Audiofield

pyproject.toml

@@ -55,7 +55,8 @@ test = [
     "python-multipart ==0.0.6",
     "fastapi >=0.92, <0.104",
     "Flask >=2.2, <2.3",
-    "Flask-SQLAlchemy >=3.0,<3.2"
+    "Flask-SQLAlchemy >=3.0,<3.2",
+    "pydub ==0.25.1"
 ]
 doc = [
     "mkdocs-material >=9.0.0, <10.0.0",

sqlalchemy_file/exceptions.py

@@ -29,3 +29,11 @@ class DimensionValidationError(ValidationError):
 
 class AspectRatioValidationError(ValidationError):
     pass
+
+
+class InvalidAudioError(ValidationError):
+    pass
+
+
+class DurationValidationError(ValidationError):
+    pass

sqlalchemy_file/types.py

@@ -9,7 +9,7 @@
 from sqlalchemy_file.mutable_list import MutableList
 from sqlalchemy_file.processors import Processor, ThumbnailGenerator
 from sqlalchemy_file.storage import StorageManager
-from sqlalchemy_file.validators import ImageValidator, Validator
+from sqlalchemy_file.validators import AudioValidator, ImageValidator, Validator
 
 
 class FileField(types.TypeDecorator):  # type: ignore
@@ -161,6 +161,58 @@ def __init__(
         )
 
 
+class AudioField(FileField):
+    """Inherits all attributes and methods from [FileField][sqlalchemy_file.types.FileField],
+    but also validates that the uploaded object is a valid audio.
+    """
+
+    cache_ok = False
+
+    def __init__(
+        self,
+        *args: Tuple[Any],
+        upload_storage: Optional[str] = None,
+        audio_validator: Optional[AudioValidator] = None,
+        validators: Optional[List[Validator]] = None,
+        processors: Optional[List[Processor]] = None,
+        upload_type: Type[File] = File,
+        multiple: Optional[bool] = False,
+        extra: Optional[Dict[str, str]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        **kwargs: Dict[str, Any],
+    ) -> None:
+        """Parameters
+        upload_storage: storage to use
+        audio_validator: AudioField use default audio
+        validator, Use this property to customize it.
+        thumbnail_size: If set, a thumbnail will be generated
+        from original image using [ThumbnailGenerator]
+        [sqlalchemy_file.processors.ThumbnailGenerator]
+        validators: List of additional validators to apply
+        processors: List of validators to apply
+        upload_type: File class to use, could be
+        used to set custom File class
+        multiple: Use this to save multiple files
+        extra: Extra attributes (driver specific).
+        """
+        if validators is None:
+            validators = []
+        if audio_validator is None:
+            audio_validator = AudioValidator()
+        validators.append(audio_validator)
+        super().__init__(
+            *args,
+            upload_storage=upload_storage,
+            validators=validators,
+            processors=processors,
+            upload_type=upload_type,
+            multiple=multiple,
+            extra=extra,
+            headers=headers,
+            **kwargs,
+        )
+
+
 class FileFieldSessionTracker:
     mapped_entities: ClassVar[Dict[Type[Any], List[str]]] = {}
 

sqlalchemy_file/validators.py

@@ -5,6 +5,8 @@
     AspectRatioValidationError,
     ContentTypeValidationError,
     DimensionValidationError,
+    DurationValidationError,
+    InvalidAudioError,
     InvalidImageError,
     SizeValidationError,
 )
@@ -226,3 +228,93 @@ def process(self, file: "File", attr_key: str) -> None:
             )
         file.update({"width": width, "height": height})
         file.original_content.seek(0)  # type: ignore[union-attr]
+
+
+class AudioValidator(ContentTypeValidator):
+    """Default Validator for AudioField. Uses `pydub` for audio operations, which might require `ffmpeg`.
+
+    Attributes:
+        min_ms: Minimum allowed duration (in milliseconds).
+        max_ms: Maximum allowed duration (in milliseconds).
+        allowed_content_types: An iterable whose items are
+                    allowed content types. Default is `audio/*`
+
+    Example:
+        ```Python
+
+        class Book(Base):
+            __tablename__ = "book"
+
+            id = Column(Integer, autoincrement=True, primary_key=True)
+            title = Column(String(100), unique=True)
+            audiobook = Column(
+                AudioField(
+                    audio_validator=AudioValidator(
+                        allowed_content_types=["audio/mpeg", "audio/mp4", "audio/wav"],
+                        min_ms=3600000, # 1 hour
+                        max_ms=43200000, # 12 hours
+                    )
+                )
+            )
+        ```
+
+    Raises:
+        ContentTypeValidationError: When file `content_type` not in allowed_content_types
+        InvalidAudioError: When file is not a valid audio
+        DimensionValidationError: When audio duration constraints fail.
+
+    Will add `width` and `height` properties to the file object
+    """
+
+    def __init__(
+        self,
+        min_ms: Optional[int] = None,
+        max_ms: Optional[int] = None,
+        allowed_content_types: Optional[List[str]] = None,
+    ):
+        audio_mime_types = [
+            "audio/aav",
+            "audio/midi",
+            "audio/x-midi",
+            "audio/mpeg",
+            "audio/ogg",
+            "audio/opus",
+            "audio/wav",
+            "audio/x-wav",
+            "audio/webm",
+            "audio/3gpp",
+        ]
+
+        super().__init__(
+            allowed_content_types
+            if allowed_content_types is not None
+            else audio_mime_types
+        )
+        self.min_duration = min_ms if min_ms else None
+        self.max_duration = max_ms if max_ms else None
+
+    def process(self, file: "File", attr_key: str) -> None:
+        super().process(file, attr_key)
+        from io import BytesIO
+
+        from pydub import AudioSegment  # type: ignore
+        from pydub.exceptions import PydubException  # type: ignore
+
+        try:
+            audio = AudioSegment.from_file(BytesIO(file.original_content.read()))  # type: ignore
+        except (PydubException, OSError):
+            raise InvalidAudioError(attr_key, "Provide valid audio file")
+        duration = int(audio.duration_seconds * 1000)
+        if self.min_duration and duration < self.min_duration:
+            raise DurationValidationError(
+                attr_key,
+                f"Minimum allowed duration is: {self.min_duration}, but {duration} is given.",
+            )
+        if self.max_duration and self.max_duration < duration:
+            raise DurationValidationError(
+                attr_key,
+                f"Maximum allowed duration is: {self.max_duration}, but {duration} is given.",
+            )
+
+        file.update({"duration": duration})
+        file.original_content.seek(0)  # type: ignore[union-attr]

tests/test_audio_field.py

@@ -0,0 +1,92 @@
+import base64
+import tempfile
+
+import pytest
+from sqlalchemy import Column, Integer, String, select
+from sqlalchemy.orm import Session, declarative_base
+from sqlalchemy_file.exceptions import ContentTypeValidationError, InvalidAudioError
+from sqlalchemy_file.storage import StorageManager
+from sqlalchemy_file.types import AudioField
+
+from tests.utils import get_test_container, get_test_engine
+
+engine = get_test_engine()
+Base = declarative_base()
+
+
+@pytest.fixture
+def fake_text_file():
+    file = tempfile.NamedTemporaryFile(suffix=".txt")
+    file.write(b"Trying to save text file as audio")
+    file.seek(0)
+    return file
+
+
+@pytest.fixture
+def fake_invalid_audio():
+    file = tempfile.NamedTemporaryFile(suffix=".mp3")
+    file.write(b"Pass through content type validation")
+    file.seek(0)
+    return file
+
+
+@pytest.fixture
+def fake_valid_audio_content():
+    return base64.b64decode(
+        "//vUxAADwAABpAAAACAAADSAAAAETEFNRQMACQgABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA//vUxAADwAABpAAAACAAADSAAAAETEFNRQMACQgABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA//vUxAADwAABpAAAACAAADSAAAAETEFNRQMACQgABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA//vUxAAAAAABpBQAACfbxBtDP+AAV4lWYsEA9mNYkf5qY8YGYo8SgKxqtMfLDqTEIAQgzN5IOMzuMMWPI+GHOizpivIWqZJYDaGYnEsRQAwjEbxS3RmyBaHK0SGc90R4KAIJgEE4zGxFNNo5Zg0zkEjTyWZMy4C4wLEhy+DeQ2ZZgiBjmEjGY0c4a1TSJlaqpGmKPt1jzqp1qvMzIYwzDSVzFpA7MMoJwzizHzJ6NVMhk3YzYUajMRMaMKQR15GX15p+FODLZKPMd0L8wNAOzELG7Md8P8zpi7jDzEiM3M6gxgDETNmLNMewKYw8i0XmVwvx37kvqGQQIQYB4KZlpEtGPsNEYlgPhgkgQhACZnaHoma8ZaYP4hBgeCImJsMGYFIIhgXgFc5vPvP+HzAtAtCANy4hgdgMo7BwABctXhgGgYGDeE8YL4GBgHgWBgAxgCAKpTf/////+WzTXbSG5G69JTo+NcZWIwCDACAJFgBxEAQtQwBwHRCAClcnR//zn////+4i7y4aA9AYrYlYHABwRFE643FxIA4qgLhgGIsAav4wGQCCYDYBAMJbhwBpgCgJGA6AGOAPf/////////////////////tPaWtcvAXIXpC1oLUZWuu+7cDxgu21f//////////yYE0wBwNZcAgAlUxGAHGwgCchALLxlshoDNDmCADEz1vFYAKYgpqKBgASEAAIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+    )
+
+
+@pytest.fixture
+def fake_valid_audio(fake_valid_audio_content):
+    file = tempfile.NamedTemporaryFile(suffix=".mp3")
+    data = fake_valid_audio_content
+    file.write(data)
+    file.seek(0)
+    return file
+
+
+class AudioBook(Base):
+    __tablename__ = "audiobook"
+
+    id = Column(Integer, autoincrement=True, primary_key=True)
+    title = Column(String(100), unique=True)
+    audio = Column(AudioField)
+
+    def __repr__(self):
+        return f"<AudioBook: id {self.id} ; name: {self.title}; audio {self.audio};>"  # pragma: no cover
+
+
+class TestAudioField:
+    def setup_method(self, method) -> None:
+        Base.metadata.create_all(engine)
+        StorageManager._clear()
+        StorageManager.add_storage("test", get_test_container("test-audio-field"))
+
+    def test_autovalidate_content_type(self, fake_text_file) -> None:
+        with Session(engine) as session:
+            session.add(AudioBook(title="Pointless Meetings", audio=fake_text_file))
+            with pytest.raises(ContentTypeValidationError):
+                session.flush()
+
+    def test_autovalidate_audio(self, fake_invalid_audio) -> None:
+        with Session(engine) as session:
+            session.add(AudioBook(title="Pointless Meetings", audio=fake_invalid_audio))
+            with pytest.raises(InvalidAudioError):
+                session.flush()
+
+    def test_create_image(self, fake_valid_audio, fake_valid_audio_content) -> None:
+        with Session(engine) as session:
+            session.add(AudioBook(title="Pointless Meetings", audio=fake_valid_audio))
+            session.flush()
+            book = session.execute(
+                select(AudioBook).where(AudioBook.title == "Pointless Meetings")
+            ).scalar_one()
+            assert book.audio.file.read() == fake_valid_audio_content
+            assert book.audio["duration"] is not None
+
+    def teardown_method(self, method):
+        for obj in StorageManager.get().list_objects():
+            obj.delete()
+        StorageManager.get().delete()
+        Base.metadata.drop_all(engine)