Source code for

Music database management.
import os

from sqlalchemy import Table, Column, Integer, String, Boolean, MetaData, \
    ForeignKey, select, create_engine
from sqlalchemy.exc import SQLAlchemyError

from pisak import dirs, logger


_MUSIC_DB_PATH = os.path.join(dirs.HOME_PISAK_DATABASES, 'music.db')

_ENGINE_URL = 'sqlite:///' + _MUSIC_DB_PATH

_LOG = logger.get_logger(__name__)

metadata = MetaData()

folders = Table('folders', metadata,
        Column('id', Integer, primary_key=True),
        Column('name', String, unique=True, nullable=False),
        Column('cover_path', String, nullable=True)

tracks = Table('tracks', metadata,
        Column('id', Integer, primary_key=True),
        Column('path', String, unique=True, nullable=False),
        Column('title', String, nullable=False),
        Column('no', Integer, nullable=True),
        Column('year', Integer, nullable=True),
        Column('cover_path', String, nullable=True),
        Column('album', String, nullable=True),
        Column('genre', String, nullable=True),
        Column('artist', String, nullable=True),
        Column('favourite', Boolean, default=False),
        Column('folder_id', Integer, ForeignKey(''), nullable=True)

engine = create_engine(_ENGINE_URL)


[docs]class DBConnector: """ Database connector. """ def __init__(self): self._conn = None def _execute(self, *args, **kwargs): try: if not self._conn: self._conn = engine.connect() return self._conn.execute(*args, **kwargs) except SQLAlchemyError as exc: _LOG.error(exc) def _close_connection(self): if self._conn: try: self._conn.close() self._conn = None except SQLAlchemyError as exc: _LOG.error(exc)
[docs] def get_folder_count(self): """ Get number of folders. :return: integer, number of folders. """ favs = 1 if self._execute(select([tracks]).where( tracks.c.favourite)).fetchone() else 0 count = len(self._execute(select([folders])).fetchall()) + favs self._close_connection() return count
[docs] def get_folders_ids(self): """ Get list of ids of all the folders from the database. :return: list of ids of all the folders, including -1 for fake favourites if there are any favourite tracks. """ ids = [row['id'] for row in self._execute(select([])).fetchall()] if self._execute(select([tracks]).where( tracks.c.favourite)).fetchone(): ids.insert(0, -1) # for fake favourites folder self._close_connection() return ids
[docs] def get_all_folders(self): """ Get all available folders. :return: list of all folders. """ folders_list = self._execute(select([folders])).fetchall() for folder in folders_list: if not self._execute(select([tracks]).where( tracks.c.folder_id == folder['id'])).fetchone(): self._execute(folders.delete().where( == folder['id'])) folders_list.remove(folder) self._include_fake_favourites_folder(folders_list) self._close_connection() return folders_list
def _include_fake_favourites_folder(self, folders_list): sample_fav = self._execute(select([tracks.c.cover_path]).where( tracks.c.favourite)).fetchone() if sample_fav: folders_list.append({'id': -1, 'name': _FAVOURITES_FOLDER_ALIAS, 'cover_path': sample_fav['cover_path']})
[docs] def get_tracks_from_folder(self, folder_id): """ Get tracks from the folder with the given index. :param folder_id: index of the folder, -1 for the favourites folder. :return: list of tracks. """ if self._is_folder(folder_id): ret = self._execute(select([tracks]).where( tracks.c.folder_id == folder_id).order_by( else: ret = self._get_favourite_tracks() self._close_connection() return ret
def _is_folder(self, folder_id): return self._execute(select([folders]).where( == folder_id)).fetchone() def _get_favourite_tracks(self): return self._execute(select([tracks]).where(tracks.c.favourite)).fetchall()
[docs] def is_track_in_favourites(self, track_path): """ Check if track with the given path is in the favourites :param track_path: path to the track. :return: True or False. """ ret = self._execute(select([tracks.c.favourite]).where( tracks.c.path == track_path)).fetchone()['favourite'] self._close_connection() return ret
[docs] def remove_track_from_favourites(self, track_path): """ Remove track with the given path from the favourites :param track_path: path to the track """ self._toggle_favourite(track_path, False)
[docs] def add_track_to_favourites(self, track_path): """ Add track with the given path to the favourites :param track_path: path to the track """ self._toggle_favourite(track_path, True)
def _toggle_favourite(self, track_path, boolean): self._execute(tracks.update().where( tracks.c.path == track_path).values(favourite=boolean)) self._close_connection()
[docs]class DBLoader(DBConnector): """ Use this to update the music library. Use `close` method when done. """ def __init__(self): super().__init__() self._collect_garbage() def _collect_garbage(self): """ Clear the db from all the non existing files. """ for track in self._execute(select([tracks])).fetchall(): if not os.path.isfile(track['path']): self._execute( tracks.delete().where( == track['id']))
[docs] def insert_many_tracks(self, tracks_list): """ Insert many tracks to the db. :param tracks_list: list of dictionaries with the tracks. """ self._execute( tracks.insert().prefix_with('OR IGNORE'), tracks_list)
[docs] def insert_folder(self, name, cover_path): """ Insert single folder to the db. :param name: name of the folder. :param cover_path: path to a cover of the given folder. :return: rowid of the inserted or already existing record. """ ret = self._execute(folders.insert().prefix_with('OR IGNORE'), name=name, cover_path=cover_path) if ret and ret.inserted_primary_key[0]: rowid = ret.inserted_primary_key[0] else: rowid = self._execute(select([]).where( == name)).fetchone()['id'] return rowid
[docs] def close(self): """ Close the db loader, close any open connections. """ self._close_connection()