# Standard library imports from typing import Optional from dataclasses import asdict import datetime as dt import json # PyQt imports # Third party imports from alchemical import Model # type: ignore from sqlalchemy import ( Boolean, DateTime, ForeignKey, String, ) from sqlalchemy.ext.associationproxy import association_proxy from sqlalchemy.engine.interfaces import Dialect from sqlalchemy.orm import ( Mapped, mapped_column, relationship, ) from sqlalchemy.orm.session import Session from sqlalchemy.types import TypeDecorator, TEXT # App imports from classes import Filter class JSONEncodedDict(TypeDecorator): """ Custom JSON Type for MariaDB (since native JSON type is just LONGTEXT) """ impl = TEXT def process_bind_param(self, value: dict | None, dialect: Dialect) -> str | None: """Convert Python dictionary to JSON string before saving.""" if value is None: return None return json.dumps(value, default=lambda o: o.__dict__) def process_result_value(self, value: str | None, dialect: Dialect) -> dict | None: """Convert JSON string back to Python dictionary after retrieval.""" if value is None: return None return json.loads(value) # Database classes class NoteColours(Model): __tablename__ = "notecolours" notecolour_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) substring: Mapped[str] = mapped_column(String(256), index=True, unique=True) colour: Mapped[str] = mapped_column(String(21), index=False) enabled: Mapped[bool] = mapped_column(default=True, index=True) foreground: Mapped[Optional[str]] = mapped_column(String(21), index=False) is_regex: Mapped[bool] = mapped_column(default=False, index=False) is_casesensitive: Mapped[bool] = mapped_column(default=False, index=False) order: Mapped[Optional[int]] = mapped_column(index=True) strip_substring: Mapped[bool] = mapped_column(default=True, index=False) def __repr__(self) -> str: return ( f"" ) def __init__( self, session: Session, substring: str, colour: str, enabled: bool = True, is_regex: bool = False, is_casesensitive: bool = False, order: Optional[int] = 0, ) -> None: self.substring = substring self.colour = colour self.enabled = enabled self.is_regex = is_regex self.is_casesensitive = is_casesensitive self.order = order session.add(self) session.commit() class Playdates(Model): __tablename__ = "playdates" playdate_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) lastplayed: Mapped[dt.datetime] = mapped_column(index=True) track_id: Mapped[int] = mapped_column( ForeignKey("tracks.track_id", ondelete="CASCADE") ) track: Mapped["Tracks"] = relationship( "Tracks", back_populates="playdates", ) def __init__( self, session: Session, track_id: int, when: dt.datetime | None = None ) -> None: """Record that track was played""" if not when: self.lastplayed = dt.datetime.now() else: self.lastplayed = when self.track_id = track_id session.add(self) session.commit() def __repr__(self) -> str: return ( f"" ) class Playlists(Model): """ Manage playlists """ __tablename__ = "playlists" playlist_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(32), unique=True) last_used: Mapped[Optional[dt.datetime]] = mapped_column(DateTime, default=None) tab: Mapped[Optional[int]] = mapped_column(default=None) open: Mapped[bool] = mapped_column(default=False) is_template: Mapped[bool] = mapped_column(default=False) rows: Mapped[list["PlaylistRows"]] = relationship( "PlaylistRows", back_populates="playlist", cascade="all, delete-orphan", order_by="PlaylistRows.row_number", ) favourite: Mapped[bool] = mapped_column( Boolean, nullable=False, index=False, default=False ) def __repr__(self) -> str: return ( f"" ) def __init__(self, session: Session, name: str, template_id: int) -> None: """Create playlist with passed name""" self.name = name self.last_used = dt.datetime.now() session.add(self) session.commit() # If a template is specified, copy from it if template_id: PlaylistRows.copy_playlist(session, template_id, self.playlist_id) class PlaylistRows(Model): __tablename__ = "playlist_rows" playlistrow_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) row_number: Mapped[int] = mapped_column(index=True) note: Mapped[str] = mapped_column( String(2048), index=False, default="", nullable=False ) playlist_id: Mapped[int] = mapped_column( ForeignKey("playlists.playlist_id", ondelete="CASCADE"), index=True ) playlist: Mapped[Playlists] = relationship(back_populates="rows") track_id: Mapped[Optional[int]] = mapped_column( ForeignKey("tracks.track_id", ondelete="CASCADE") ) track: Mapped["Tracks"] = relationship( "Tracks", back_populates="playlistrows", ) played: Mapped[bool] = mapped_column( Boolean, nullable=False, index=False, default=False ) def __repr__(self) -> str: return ( f"" ) def __init__( self, session: Session, playlist_id: int, row_number: int, note: str = "", track_id: Optional[int] = None, ) -> None: """Create PlaylistRows object""" self.playlist_id = playlist_id self.track_id = track_id self.row_number = row_number self.note = note session.add(self) session.commit() class Queries(Model): __tablename__ = "queries" query_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(128), nullable=False) _filter_data: Mapped[dict | None] = mapped_column( "filter_data", JSONEncodedDict, nullable=False ) favourite: Mapped[bool] = mapped_column( Boolean, nullable=False, index=False, default=False ) def _get_filter(self) -> Filter: """Convert stored JSON dictionary to a Filter object.""" if isinstance(self._filter_data, dict): return Filter(**self._filter_data) return Filter() # Default object if None or invalid data def _set_filter(self, value: Filter | None) -> None: """Convert a Filter object to JSON before storing.""" self._filter_data = asdict(value) if isinstance(value, Filter) else None # Single definition of `filter` filter = property(_get_filter, _set_filter) def __repr__(self) -> str: return f"" def __init__( self, session: Session, name: str, filter: Filter, favourite: bool = False, ) -> None: """Create new query""" self.name = name self.filter = filter self.favourite = favourite session.add(self) session.commit() class Settings(Model): """Manage settings""" __tablename__ = "settings" setting_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(64), unique=True) f_datetime: Mapped[Optional[dt.datetime]] = mapped_column(default=None) f_int: Mapped[Optional[int]] = mapped_column(default=None) f_string: Mapped[Optional[str]] = mapped_column(String(128), default=None) def __repr__(self) -> str: return ( f"" ) def __init__(self, session: Session, name: str) -> None: self.name = name session.add(self) session.commit() class Tracks(Model): __tablename__ = "tracks" track_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) artist: Mapped[str] = mapped_column(String(256), index=True) bitrate: Mapped[int] = mapped_column(default=None) duration: Mapped[int] = mapped_column(index=True) fade_at: Mapped[int] = mapped_column(index=False) intro: Mapped[Optional[int]] = mapped_column(default=None) path: Mapped[str] = mapped_column(String(2048), index=False, unique=True) silence_at: Mapped[int] = mapped_column(index=False) start_gap: Mapped[int] = mapped_column(index=False) title: Mapped[str] = mapped_column(String(256), index=True) playlistrows: Mapped[list[PlaylistRows]] = relationship( "PlaylistRows", back_populates="track", cascade="all, delete-orphan", ) playlists = association_proxy("playlistrows", "playlist") playdates: Mapped[list[Playdates]] = relationship( "Playdates", back_populates="track", cascade="all, delete-orphan", lazy="joined", ) def __repr__(self) -> str: return ( f"" ) def __init__( self, session: Session, path: str, title: str, artist: str, duration: int, start_gap: int, fade_at: int, silence_at: int, bitrate: int, ) -> None: self.path = path self.title = title self.artist = artist self.bitrate = bitrate self.duration = duration self.start_gap = start_gap self.fade_at = fade_at self.silence_at = silence_at session.add(self) session.commit()