db: Add a Library Table
This table allows us to work with Library playlists that are represented by a filesystem path. The user can manually enable or disable library paths to prevent their tracks from showing up in the Collection playlist. Additionally, library paths have an online property to determine if the library still exists in the filesystem to prevent us from removing tracks due to a broken NFS mount or symlink. Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
parent
b7f1a05967
commit
24c1a31367
|
|
@ -8,6 +8,7 @@ from . import artists
|
|||
from . import connection
|
||||
from . import decades
|
||||
from . import genres
|
||||
from . import libraries
|
||||
from . import playlist
|
||||
from . import media
|
||||
from . import playlists
|
||||
|
|
@ -41,6 +42,7 @@ class Connection(connection.Connection):
|
|||
self.genres = genres.Table(self)
|
||||
self.decades = decades.Table(self)
|
||||
self.years = years.Table(self, queue=self.decades.queue)
|
||||
self.libraries = libraries.Table(self)
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the database connection."""
|
||||
|
|
@ -64,7 +66,7 @@ class Connection(connection.Connection):
|
|||
def playlist_tables(self) -> Generator[playlist.Table, None, None]:
|
||||
"""Iterate over each playlist table."""
|
||||
for tbl in [self.playlists, self.artists, self.albums, self.media,
|
||||
self.genres, self.decades, self.years]:
|
||||
self.genres, self.decades, self.years, self.libraries]:
|
||||
yield tbl
|
||||
|
||||
def set_active_playlist(self, plist: playlist.Playlist) -> None:
|
||||
|
|
|
|||
|
|
@ -305,6 +305,40 @@ CREATE TRIGGER years_delete_trigger AFTER DELETE ON years
|
|||
END;
|
||||
|
||||
|
||||
/*******************************
|
||||
* *
|
||||
* Library Paths *
|
||||
* *
|
||||
*******************************/
|
||||
|
||||
CREATE TABLE libraries (
|
||||
libraryid INTEGER PRIMARY KEY,
|
||||
propertyid INTEGER REFERENCES playlist_properties (propertyid)
|
||||
ON DELETE CASCADE
|
||||
ON UPDATE CASCADE,
|
||||
path PATH UNIQUE,
|
||||
enabled BOOLEAN DEFAULT TRUE,
|
||||
deleting BOOLEAN DEFAULT FALSE
|
||||
);
|
||||
|
||||
CREATE VIEW libraries_view AS
|
||||
SELECT libraryid, propertyid, path, path as name, enabled, active
|
||||
FROM libraries
|
||||
JOIN playlist_properties USING (propertyid);
|
||||
|
||||
CREATE TRIGGER libraries_insert_trigger AFTER INSERT ON libraries
|
||||
BEGIN
|
||||
INSERT INTO playlist_properties (active) VALUES (False);
|
||||
UPDATE libraries SET propertyid = last_insert_rowid()
|
||||
WHERE libraryid = NEW.libraryid;
|
||||
END;
|
||||
|
||||
CREATE TRIGGER libraries_delete_trigger AFTER DELETE ON libraries
|
||||
BEGIN
|
||||
DELETE FROM playlist_properties WHERE propertyid = OLD.propertyid;
|
||||
END;
|
||||
|
||||
|
||||
/******************************************
|
||||
* *
|
||||
* Create Default Playlists *
|
||||
|
|
|
|||
|
|
@ -0,0 +1,93 @@
|
|||
# Copyright 2022 (c) Anna Schumaker
|
||||
"""A custom Gio.ListModel for working with libraries."""
|
||||
import pathlib
|
||||
import sqlite3
|
||||
from gi.repository import GObject
|
||||
from . import idle
|
||||
from . import playlist
|
||||
|
||||
|
||||
class Library(playlist.Playlist):
|
||||
"""Our custom Library with path and enabled properties."""
|
||||
|
||||
libraryid = GObject.Property(type=int)
|
||||
path = GObject.Property(type=GObject.TYPE_PYOBJECT)
|
||||
enabled = GObject.Property(type=bool, default=True)
|
||||
deleting = GObject.Property(type=bool, default=False)
|
||||
|
||||
queue = GObject.Property(type=idle.Queue)
|
||||
online = GObject.Property(type=bool, default=False)
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
"""Initialize our Library object."""
|
||||
super().__init__(queue=idle.Queue(), **kwargs)
|
||||
|
||||
def __queue_delete(self) -> bool:
|
||||
self.table.delete(self)
|
||||
return True
|
||||
|
||||
def do_update(self, column: str) -> bool:
|
||||
"""Update a Library playlist."""
|
||||
match column:
|
||||
case "online": self.table.notify_online(self)
|
||||
case _: return super().do_update(column)
|
||||
return True
|
||||
|
||||
def delete(self) -> bool:
|
||||
"""Delete this Library."""
|
||||
if self.deleting is False:
|
||||
self.deleting = True
|
||||
self.queue.push(self.__queue_delete)
|
||||
return True
|
||||
return False
|
||||
|
||||
@property
|
||||
def primary_key(self) -> int:
|
||||
"""Get this library's primary key."""
|
||||
return self.libraryid
|
||||
|
||||
|
||||
class Table(playlist.Table):
|
||||
"""Our Library ListModel."""
|
||||
|
||||
def do_construct(self, **kwargs) -> Library:
|
||||
"""Construct a new library."""
|
||||
return Library(**kwargs)
|
||||
|
||||
def do_sql_delete(self, library: Library) -> sqlite3.Cursor:
|
||||
"""Delete a library."""
|
||||
return self.sql("DELETE FROM libraries WHERE libraryid=?",
|
||||
library.libraryid)
|
||||
|
||||
def do_sql_insert(self, path: pathlib.Path) -> sqlite3.Cursor:
|
||||
"""Create a new library."""
|
||||
if cur := self.sql("INSERT INTO libraries (path) VALUES (?)", path):
|
||||
return self.sql("SELECT * FROM libraries_view WHERE libraryid=?",
|
||||
cur.lastrowid)
|
||||
|
||||
def do_sql_glob(self, glob: str) -> sqlite3.Cursor:
|
||||
"""Search for libraries matching the search text."""
|
||||
return self.sql("""SELECT libraryid FROM libraries_view
|
||||
WHERE name GLOB ?""", glob)
|
||||
|
||||
def do_sql_select_all(self) -> sqlite3.Cursor:
|
||||
"""Load libraries from the database."""
|
||||
return self.sql("SELECT * FROM libraries_view")
|
||||
|
||||
def do_sql_select_one(self, path: pathlib.Path) -> sqlite3.Cursor:
|
||||
"""Look up a library by path."""
|
||||
return self.sql("SELECT libraryid FROM libraries WHERE path=?", path)
|
||||
|
||||
def do_sql_update(self, library: Library, column: str, newval) -> bool:
|
||||
"""Update a Library playlist."""
|
||||
return self.sql(f"UPDATE libraries SET {column}=? WHERE rowid=?",
|
||||
newval, library.libraryid)
|
||||
|
||||
def notify_online(self, library: Library) -> None:
|
||||
"""Notify that a library's online status has changed."""
|
||||
if not library.online or self.loaded:
|
||||
self.emit("library-online", library)
|
||||
|
||||
@GObject.Signal(arg_types=(Library,))
|
||||
def library_online(self, library: Library) -> None:
|
||||
"""Signal that a library online status has changed."""
|
||||
|
|
@ -47,6 +47,7 @@ class TestConnection(tests.util.TestCase):
|
|||
self.assertIsInstance(self.sql.genres, emmental.db.genres.Table)
|
||||
self.assertIsInstance(self.sql.decades, emmental.db.decades.Table)
|
||||
self.assertIsInstance(self.sql.years, emmental.db.years.Table)
|
||||
self.assertIsInstance(self.sql.libraries, emmental.db.libraries.Table)
|
||||
|
||||
self.assertEqual(self.sql.albums.queue, self.sql.artists.queue)
|
||||
self.assertEqual(self.sql.media.queue, self.sql.artists.queue)
|
||||
|
|
@ -56,7 +57,7 @@ class TestConnection(tests.util.TestCase):
|
|||
[self.sql.playlists, self.sql.artists,
|
||||
self.sql.albums, self.sql.media,
|
||||
self.sql.genres, self.sql.decades,
|
||||
self.sql.years])
|
||||
self.sql.years, self.sql.libraries])
|
||||
|
||||
def test_load(self):
|
||||
"""Check that calling load() loads the tables."""
|
||||
|
|
|
|||
|
|
@ -0,0 +1,186 @@
|
|||
# Copyright 2022 (c) Anna Schumaker
|
||||
"""Tests our library Gio.ListModel."""
|
||||
import pathlib
|
||||
import emmental.db
|
||||
import tests.util
|
||||
import unittest.mock
|
||||
|
||||
|
||||
class TestLibraryObject(tests.util.TestCase):
|
||||
"""Tests our library object."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up common variables."""
|
||||
super().setUp()
|
||||
self.table = self.sql.libraries
|
||||
self.path = pathlib.Path("/a/b/c")
|
||||
self.library = emmental.db.libraries.Library(table=self.table,
|
||||
libraryid=123,
|
||||
propertyid=456,
|
||||
path=self.path,
|
||||
name=str(self.path))
|
||||
|
||||
def test_init(self):
|
||||
"""Test that the Library is set up properly."""
|
||||
self.assertIsInstance(self.library, emmental.db.playlist.Playlist)
|
||||
self.assertIsInstance(self.library.queue, emmental.db.idle.Queue)
|
||||
|
||||
self.assertEqual(self.library.table, self.table)
|
||||
self.assertEqual(self.library.propertyid, 456)
|
||||
self.assertEqual(self.library.libraryid, 123)
|
||||
self.assertEqual(self.library.primary_key, 123)
|
||||
self.assertEqual(self.library.path, self.path)
|
||||
self.assertTrue(self.library.enabled)
|
||||
self.assertFalse(self.library.deleting)
|
||||
self.assertFalse(self.library.online)
|
||||
self.assertIsNone(self.library.parent)
|
||||
|
||||
def test_delete(self):
|
||||
"""Test deleting a Library path."""
|
||||
with unittest.mock.patch.object(self.table, "delete") as mock_delete:
|
||||
with unittest.mock.patch.object(self.table, "update"):
|
||||
self.assertTrue(self.library.delete())
|
||||
self.assertTrue(self.library.deleting)
|
||||
mock_delete.assert_not_called()
|
||||
self.assertEqual(self.library.queue[0],
|
||||
(self.library._Library__queue_delete,))
|
||||
|
||||
self.assertFalse(self.library.delete())
|
||||
self.assertEqual(self.library.queue.total, 1)
|
||||
|
||||
self.library.queue.complete()
|
||||
mock_delete.assert_called_with(self.library)
|
||||
|
||||
def test_online(self):
|
||||
"""Test that changing the online property notifies the table."""
|
||||
with unittest.mock.patch.object(self.table,
|
||||
"notify_online") as mock_notify:
|
||||
self.library.online = True
|
||||
mock_notify.assert_called_with(self.library)
|
||||
|
||||
|
||||
class TestLibraryTable(tests.util.TestCase):
|
||||
"""Tests our library table."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up common variables."""
|
||||
tests.util.TestCase.setUp(self)
|
||||
self.table = self.sql.libraries
|
||||
|
||||
def test_init(self):
|
||||
"""Test that the library model is configured correctly."""
|
||||
self.assertIsInstance(self.table, emmental.db.playlist.Table)
|
||||
self.assertEqual(len(self.table), 0)
|
||||
|
||||
def test_construct(self):
|
||||
"""Test constructing a new library."""
|
||||
library = self.table.construct(propertyid=1, libraryid=1,
|
||||
path=pathlib.Path("/a/b/c"),
|
||||
name="/a/b/c")
|
||||
self.assertIsInstance(library, emmental.db.libraries.Library)
|
||||
self.assertEqual(library.table, self.table)
|
||||
self.assertEqual(library.propertyid, 1)
|
||||
self.assertEqual(library.libraryid, 1)
|
||||
self.assertEqual(library.path, pathlib.Path("/a/b/c"))
|
||||
self.assertEqual(library.name, "/a/b/c")
|
||||
self.assertFalse(library.active)
|
||||
|
||||
def test_create(self):
|
||||
"""Test creating a library."""
|
||||
library = self.table.create(pathlib.Path("/a/b/c"))
|
||||
self.assertIsInstance(library, emmental.db.libraries.Library)
|
||||
self.assertEqual(library.path, pathlib.Path("/a/b/c"))
|
||||
|
||||
cur = self.sql("SELECT COUNT(path) FROM libraries")
|
||||
self.assertEqual(cur.fetchone()["COUNT(path)"], 1)
|
||||
self.assertEqual(len(self.table.store), 1)
|
||||
self.assertEqual(self.table.store.get_item(0), library)
|
||||
|
||||
cur = self.sql("""SELECT COUNT(*) FROM playlist_properties
|
||||
WHERE propertyid=?""", library.propertyid)
|
||||
self.assertEqual(cur.fetchone()["COUNT(*)"], 1)
|
||||
|
||||
self.assertIsNone(self.table.create("/a/b/c"))
|
||||
|
||||
def test_delete(self):
|
||||
"""Test deleting a library."""
|
||||
library = self.table.create(pathlib.Path("/a/b/c"))
|
||||
self.assertTrue(self.table.delete(library))
|
||||
self.assertIsNone(self.table.index(library))
|
||||
|
||||
cur = self.sql("SELECT COUNT(path) FROM libraries")
|
||||
self.assertEqual(cur.fetchone()["COUNT(path)"], 0)
|
||||
self.assertEqual(len(self.table), 0)
|
||||
self.assertIsNone(self.table.get_item(0))
|
||||
|
||||
cur = self.sql("""SELECT COUNT(*) FROM playlist_properties
|
||||
WHERE propertyid=?""", library.propertyid)
|
||||
self.assertEqual(cur.fetchone()["COUNT(*)"], 0)
|
||||
|
||||
self.assertFalse(self.table.delete(library))
|
||||
|
||||
def test_filter(self):
|
||||
"""Test filtering the library model."""
|
||||
self.table.create(pathlib.Path("/a/b/c"))
|
||||
self.table.create(pathlib.Path("/a/b/d"))
|
||||
|
||||
self.table.filter("*c", now=True)
|
||||
self.assertSetEqual(self.table.get_filter().keys, {1})
|
||||
self.table.filter("*a/b*", now=True)
|
||||
self.assertSetEqual(self.table.get_filter().keys, {1, 2})
|
||||
|
||||
def test_load(self):
|
||||
"""Test loading libraries from the database."""
|
||||
self.table.create("/a/b/c")
|
||||
self.table.create("/a/b/d").enabled = False
|
||||
|
||||
libraries2 = emmental.db.libraries.Table(self.sql)
|
||||
self.assertEqual(len(libraries2), 0)
|
||||
|
||||
libraries2.load(now=True)
|
||||
self.assertEqual(len(libraries2), 2)
|
||||
|
||||
self.assertEqual(libraries2.get_item(0).libraryid, 1)
|
||||
self.assertEqual(libraries2.get_item(0).path, pathlib.Path("/a/b/c"))
|
||||
self.assertTrue(libraries2.get_item(0).enabled)
|
||||
|
||||
self.assertEqual(libraries2.get_item(1).libraryid, 2)
|
||||
self.assertEqual(libraries2.get_item(1).path, pathlib.Path("/a/b/d"))
|
||||
self.assertFalse(libraries2.get_item(1).enabled)
|
||||
|
||||
def test_lookup(self):
|
||||
"""Test looking up a library."""
|
||||
library = self.table.create(pathlib.Path("/a/b/c"))
|
||||
self.assertEqual(self.table.lookup(pathlib.Path("/a/b/c/")), library)
|
||||
self.assertIsNone(self.table.lookup(pathlib.Path("/no/library/path")))
|
||||
|
||||
def test_update(self):
|
||||
"""Test updating genre attributes."""
|
||||
library = self.table.create("/a/b/c")
|
||||
library.active = True
|
||||
library.enabled = False
|
||||
|
||||
row = self.sql("""SELECT active, enabled FROM libraries_view
|
||||
WHERE libraryid=?""", library.libraryid).fetchone()
|
||||
self.assertTrue(row["active"])
|
||||
self.assertFalse(row["enabled"])
|
||||
|
||||
def test_library_online(self):
|
||||
"""Test the library-online signal."""
|
||||
library = self.table.create(pathlib.Path("/a/b/c"))
|
||||
callback = unittest.mock.Mock()
|
||||
self.table.connect("library-online", callback)
|
||||
|
||||
library.online = True
|
||||
callback.assert_not_called()
|
||||
library.online = False
|
||||
callback.assert_called_with(self.table, library)
|
||||
|
||||
callback.reset_mock()
|
||||
self.table.loaded = True
|
||||
library.online = True
|
||||
callback.assert_called_with(self.table, library)
|
||||
|
||||
callback.reset_mock()
|
||||
library.online = False
|
||||
callback.assert_called_with(self.table, library)
|
||||
Loading…
Reference in New Issue