From fac4fdf2123495188fb6d9f96473fcdf1b37ad70 Mon Sep 17 00:00:00 2001 From: JSC Date: Sat, 5 Jul 2025 18:31:47 +0200 Subject: [PATCH] feat(stream): add Stream model for managing streaming service links to sounds; update Sound model to include relationship with Stream --- app/models/__init__.py | 3 +- app/models/sound.py | 6 ++ app/models/stream.py | 169 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 177 insertions(+), 1 deletion(-) create mode 100644 app/models/stream.py diff --git a/app/models/__init__.py b/app/models/__init__.py index 23b6558..d1eb59b 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -4,7 +4,8 @@ from .plan import Plan from .playlist import Playlist from .playlist_sound import PlaylistSound from .sound import Sound +from .stream import Stream from .user import User from .user_oauth import UserOAuth -__all__ = ["Plan", "Playlist", "PlaylistSound", "Sound", "User", "UserOAuth"] +__all__ = ["Plan", "Playlist", "PlaylistSound", "Sound", "Stream", "User", "UserOAuth"] diff --git a/app/models/sound.py b/app/models/sound.py index b23539a..f207083 100644 --- a/app/models/sound.py +++ b/app/models/sound.py @@ -12,6 +12,7 @@ from app.database import db if TYPE_CHECKING: from app.models.playlist_sound import PlaylistSound + from app.models.stream import Stream class SoundType(Enum): @@ -96,6 +97,11 @@ class Sound(db.Model): back_populates="sound", cascade="all, delete-orphan", ) + streams: Mapped[list["Stream"]] = relationship( + "Stream", + back_populates="sound", + cascade="all, delete-orphan", + ) def __repr__(self) -> str: """String representation of Sound.""" diff --git a/app/models/stream.py b/app/models/stream.py new file mode 100644 index 0000000..b9137c6 --- /dev/null +++ b/app/models/stream.py @@ -0,0 +1,169 @@ +"""Stream model for storing streaming service links to sounds.""" + +from datetime import datetime +from typing import TYPE_CHECKING, Optional +from zoneinfo import ZoneInfo + +from sqlalchemy import DateTime, ForeignKey, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.database import db + +if TYPE_CHECKING: + from app.models.sound import Sound + + +class Stream(db.Model): + """Model for storing streaming service information linked to sounds.""" + + __tablename__ = "stream" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + service: Mapped[str] = mapped_column(String(50), nullable=False) + service_id: Mapped[str] = mapped_column(String(255), nullable=False) + sound_id: Mapped[int] = mapped_column( + Integer, ForeignKey("sound.id"), nullable=False + ) + url: Mapped[str] = mapped_column(Text, nullable=False) + title: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) + track: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) + artist: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) + album: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) + genre: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) + status: Mapped[str] = mapped_column(String(50), nullable=False, default="active") + created_at: Mapped[datetime] = mapped_column( + DateTime, + default=lambda: datetime.now(tz=ZoneInfo("UTC")), + nullable=False, + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime, + default=lambda: datetime.now(tz=ZoneInfo("UTC")), + onupdate=lambda: datetime.now(tz=ZoneInfo("UTC")), + nullable=False, + ) + + # Relationships + sound: Mapped["Sound"] = relationship("Sound", back_populates="streams") + + def __repr__(self) -> str: + """String representation of the stream.""" + return f"" + + def to_dict(self) -> dict: + """Convert stream to dictionary representation.""" + return { + "id": self.id, + "service": self.service, + "service_id": self.service_id, + "sound_id": self.sound_id, + "url": self.url, + "title": self.title, + "track": self.track, + "artist": self.artist, + "album": self.album, + "genre": self.genre, + "status": self.status, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } + + @classmethod + def create_stream( + cls, + service: str, + service_id: str, + sound_id: int, + url: str, + title: Optional[str] = None, + track: Optional[str] = None, + artist: Optional[str] = None, + album: Optional[str] = None, + genre: Optional[str] = None, + status: str = "active", + commit: bool = True, + ) -> "Stream": + """Create a new stream record.""" + stream = cls( + service=service, + service_id=service_id, + sound_id=sound_id, + url=url, + title=title, + track=track, + artist=artist, + album=album, + genre=genre, + status=status, + ) + + db.session.add(stream) + if commit: + db.session.commit() + + return stream + + @classmethod + def find_by_service_and_id( + cls, service: str, service_id: str + ) -> Optional["Stream"]: + """Find stream by service and service_id.""" + return cls.query.filter_by(service=service, service_id=service_id).first() + + @classmethod + def find_by_sound(cls, sound_id: int) -> list["Stream"]: + """Find all streams for a specific sound.""" + return cls.query.filter_by(sound_id=sound_id).all() + + @classmethod + def find_by_service(cls, service: str) -> list["Stream"]: + """Find all streams for a specific service.""" + return cls.query.filter_by(service=service).all() + + @classmethod + def find_by_status(cls, status: str) -> list["Stream"]: + """Find all streams with a specific status.""" + return cls.query.filter_by(status=status).all() + + @classmethod + def find_active_streams(cls) -> list["Stream"]: + """Find all active streams.""" + return cls.query.filter_by(status="active").all() + + def update_metadata( + self, + title: Optional[str] = None, + track: Optional[str] = None, + artist: Optional[str] = None, + album: Optional[str] = None, + genre: Optional[str] = None, + commit: bool = True, + ) -> None: + """Update stream metadata.""" + if title is not None: + self.title = title + if track is not None: + self.track = track + if artist is not None: + self.artist = artist + if album is not None: + self.album = album + if genre is not None: + self.genre = genre + + if commit: + db.session.commit() + + def set_status(self, status: str, commit: bool = True) -> None: + """Update stream status.""" + self.status = status + if commit: + db.session.commit() + + def is_active(self) -> bool: + """Check if stream is active.""" + return self.status == "active" + + def get_display_name(self) -> str: + """Get a display name for the stream (title or track or service_id).""" + return self.title or self.track or self.service_id \ No newline at end of file