83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
# Copyright 2022 (c) Anna Schumaker
|
|
"""A custom Gio.ListModel for working with albums."""
|
|
import pathlib
|
|
import sqlite3
|
|
from gi.repository import GObject
|
|
from .. import format
|
|
from . import playlist
|
|
|
|
|
|
class Album(playlist.Playlist):
|
|
"""Our custom Album with a ListModel representing mediums."""
|
|
|
|
albumid = GObject.Property(type=int)
|
|
artist = GObject.Property(type=str)
|
|
release = GObject.Property(type=str)
|
|
mbid = GObject.Property(type=str)
|
|
cover = GObject.Property(type=GObject.TYPE_PYOBJECT)
|
|
|
|
@property
|
|
def primary_key(self) -> int:
|
|
"""Get the Album primary key."""
|
|
return self.albumid
|
|
|
|
|
|
class Table(playlist.Table):
|
|
"""Our Album Table."""
|
|
|
|
def do_construct(self, **kwargs) -> Album:
|
|
"""Construct a new album."""
|
|
return Album(**kwargs)
|
|
|
|
def do_get_sort_key(self, album: Album) -> tuple[tuple, bool,
|
|
str, tuple, str]:
|
|
"""Get a sort key for the requested Artist."""
|
|
return (format.sort_key(album.name),
|
|
len(album.mbid) == 0, album.mbid.casefold(),
|
|
format.sort_key(album.artist),
|
|
album.release)
|
|
|
|
def do_sql_delete(self, album: Album) -> sqlite3.Cursor:
|
|
"""Delete an album."""
|
|
return self.sql("DELETE FROM albums WHERE albumid=?", album.albumid)
|
|
|
|
def do_sql_glob(self, glob: str) -> sqlite3.Cursor:
|
|
"""Search for albums matching the search text."""
|
|
return self.sql("""SELECT albumid FROM albums
|
|
WHERE CASEFOLD(name) GLOB ?""", glob)
|
|
|
|
def do_sql_insert(self, name: str, artist: str,
|
|
release: str, *, mbid: str = "",
|
|
cover: pathlib.Path = None) -> sqlite3.Cursor | None:
|
|
"""Create a new album."""
|
|
if cur := self.sql("""INSERT INTO albums
|
|
(name, artist, release, mbid, cover)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
name, artist, release, mbid, cover):
|
|
return self.sql("SELECT * FROM albums_view WHERE albumid=?",
|
|
cur.lastrowid)
|
|
|
|
def do_sql_select_all(self) -> sqlite3.Cursor:
|
|
"""Load albums from the database."""
|
|
return self.sql("SELECT * FROM albums_view")
|
|
|
|
def do_sql_select_one(self, name: str = None,
|
|
artist: str = None, release: str = None,
|
|
*, mbid: str = "") -> sqlite3.Cursor:
|
|
"""Look up an albums by name, mbid, artist, and release."""
|
|
where = ["mbid=?"]
|
|
args = [mbid.lower()]
|
|
|
|
if None not in (name, artist, release):
|
|
where.extend(["CASEFOLD(name)=?",
|
|
"CASEFOLD(artist)=?", "release=?"])
|
|
args.extend([name.casefold(), artist.casefold(), release])
|
|
|
|
return self.sql(f"""SELECT albumid FROM albums
|
|
WHERE {" AND ".join(where)}""", *args)
|
|
|
|
def do_sql_update(self, album: Album, column: str, newval) -> bool:
|
|
"""Rename an album."""
|
|
return self.sql(f"UPDATE albums SET {column}=? WHERE albumid=?",
|
|
newval, album.albumid)
|