125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""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,
|
|
UniqueConstraint,
|
|
)
|
|
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=True)
|
|
service_id: Mapped[str] = mapped_column(String(255), nullable=True)
|
|
sound_id: Mapped[int] = mapped_column(
|
|
Integer, ForeignKey("sound.id"), nullable=True
|
|
)
|
|
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="pending"
|
|
)
|
|
error: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
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")
|
|
|
|
# Constraints
|
|
__table_args__ = (
|
|
UniqueConstraint("service", "service_id", name="unique_service_stream"),
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
"""String representation of the stream."""
|
|
return f"<Stream(id={self.id}, service='{self.service}', service_id='{self.service_id}', sound_id={self.sound_id})>"
|
|
|
|
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,
|
|
"error": self.error,
|
|
"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,
|
|
url: str,
|
|
service: Optional[str] = None,
|
|
service_id: Optional[str] = None,
|
|
sound_id: Optional[int] = None,
|
|
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
|