musicmuster/app/model.py

#!/usr/bin/python3

import os
import sqlalchemy

from datetime import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import (
    Column,
    DateTime,
    Float,
    ForeignKey,
    Integer,
    String,
    Table
)
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.orm import relationship, sessionmaker

from config import Config

from log import DEBUG, ERROR, INFO

INFO("Starting")

# Create session at the global level as per
# https://docs.sqlalchemy.org/en/13/orm/session_basics.html

# Set up database connection
INFO("Connect to database")
engine = sqlalchemy.create_engine(f"{Config.MYSQL_CONNECT}?charset=utf8",
                                  encoding='utf-8',
                                  echo=Config.DISPLAY_SQL)
Base = declarative_base()
Base.metadata.create_all(engine)

# Create a Session
Session = sessionmaker(bind=engine)
session = Session()


# Database classes
class Settings(Base):
    __tablename__ = 'settings'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, unique=True)
    f_datetime = Column(DateTime, default=None, nullable=True)
    f_int = Column(Integer, default=None, nullable=True)
    f_string = Column(String(128), default=None, nullable=True)

    @classmethod
    def get_int(cls, name):
        try:
            int_setting = session.query(cls).filter(
                cls.name == name).one()
        except NoResultFound:
            int_setting = Settings()
            int_setting.name = name
            int_setting.f_int = None
            session.add(int_setting)
            session.commit()
        return int_setting

    def update(self, data):
        for key, value in data.items():
            assert hasattr(self, key)
            setattr(self, key, value)
        session.commit()


playlist_tracks = Table(
    'playlistracks',
    Base.metadata,
    Column('playlist_id', Integer, ForeignKey('playlists.id')),
    Column('track_id', Integer, ForeignKey('tracks.id'))
)


class Playlists(Base):
    """
    Usage:

    In [3]: pl = session.query(Playlists).filter(Playlists.id == 1).one()

    In [4]: pl
    Out[4]: <Playlist(id=1, name=Default>

    In [5]: tr = session.query(Tracks).filter(Tracks.id == 3837).one()

    In [6]: tr
    Out[6]: <Track(id=3837, title=Babe, artist=Various, path=/home/[...]

    In [7]: pl.tracks.append(tr)
       ...: session.commit()

    In [8]: tr.playlists
    Out[8]: [<Playlist(id=1, name=Default>]

    In [9]: pl.tracks
    Out[9]: [<Track(id=3837, title=Babe, artist=Various, path=/home/[...]
    """

    __tablename__ = "playlists"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, unique=True)
    tracks = relationship(
        "Tracks",
        secondary=playlist_tracks,
        back_populates="playlists")

    def __repr__(self):
        return (f"<Playlist(id={self.id}, name={self.name}>")

    # Currently we only support one playlist, so make that obvious from
    # function name
    @classmethod
    def get_only_playlist(cls):
        return session.query(Playlists).filter(Playlists.id == 1).one()

    def add_track(self, track):
        self.tracks.append(track)

    def get_tracks(self):
        return self.tracks


class Tracks(Base):
    __tablename__ = 'tracks'

    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(256), index=True)
    artist = Column(String(256), index=True)
    duration = Column(Integer, index=True)
    start_gap = Column(Integer, index=False)
    fade_at = Column(Integer, index=False)
    silence_at = Column(Integer, index=False)
    path = Column(String(2048), index=False, nullable=False)
    mtime = Column(Float, index=True)
    lastplayed = Column(DateTime, index=True, default=None)
    playlists = relationship(
        "Playlists",
        secondary=playlist_tracks,
        back_populates="tracks")
    playdates_id = Column(Integer, ForeignKey('playdates.id'))
    playdates = relationship("Playdates", back_populates="tracks")

    def __repr__(self):
        return (
            f"<Track(id={self.id}, title={self.title}, "
            f"artist={self.artist}, path={self.path}>"
        )

    @classmethod
    def get_or_create(cls, path):
        DEBUG(f"get_or_create(cls, {path})")
        try:
            track = session.query(cls).filter(cls.path == path).one()
        except NoResultFound:
            track = Tracks()
            track.path = path
            session.add(track)
        return track

    @staticmethod
    def get_path(id):
        try:
            path = session.query(Tracks.path).filter(Tracks.id == id).one()[0]
            return os.path.join(Config.ROOT, path)
        except NoResultFound:
            print(f"Can't find track id {id}")
            return None

    @staticmethod
    def get_track(id):
        try:
            DEBUG(f"get_track({id})")
            track = session.query(Tracks).filter(Tracks.id == id).one()
            return track
        except NoResultFound:
            ERROR(f"get_track({id}): not found")
            return None

    @staticmethod
    def search_titles(text):
        return (
            session.query(Tracks)
            .filter(Tracks.title.ilike(f"%{text}%"))
            .order_by(Tracks.title)
        ).all()

    @staticmethod
    def track_from_id(id):
        return session.query(Tracks).filter(
            Tracks.id == id).one()


class Playdates(Base):
    __tablename__ = 'playdates'

    id = Column(Integer, primary_key=True, autoincrement=True)
    lastplayed = Column(DateTime, index=True, default=None)
    tracks = relationship("Tracks", back_populates="playdates")

    @staticmethod
    def add_playdate(track):
        DEBUG(f"add_playdate({track})")
        pd = Playdates()
        pd.lastplayed = datetime.now()
        pd.tracks.append(track)
        session.add(pd)
        session.commit()