anna/emmental
anna
/
emmental
1
0
Fork 0
Code Issues 11 Pull Requests Projects 1 Releases 12 Wiki Activity

db: Remove the PlaylistMap objects 

We can now do everything it did and more through the various user
playlist objects.

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
Anna Schumaker 2021-10-17 22:04:22 -04:00
parent bba00b3d27
commit 932663f872
4 changed files with 18 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
db/__init__.py
View File

@ -28,11 +28,6 @@ def make_fake_track(trackno, length, title, path, lib="/a/b/c", art="Test Artist
def reset():
mods = [ track, state, user, artist, album,
disc, genre, decade, year, library ]
for mod in mods: mod.Table.reset()
user.Map.reset()
user.TempMap.reset()
if lib.version.TESTING: reset()

78
db/test_user.py
View File

@ -227,81 +227,3 @@ class TestUserTable(unittest.TestCase):
self.assertEqual(table.get_item(2).name, "New Tracks")
self.assertEqual(table.get_item(3).name, "Previous")
self.assertEqual(table.get_item(4).name, "Queued Tracks")
class TestPlaylistMap(unittest.TestCase):
def setUp(self):
db.reset()
def test_playlist_map_init(self):
self.assertIsInstance(db.user.Map, db.user.PlaylistMap)
self.assertIsInstance(db.user.TempMap, db.user.PlaylistMap)
self.assertEqual(db.user.Map.map_lhs, db.user.Table.get)
self.assertEqual(db.user.Map.map_rhs, db.user.Map.get_track)
self.assertFalse(db.user.Map.temporary)
self.assertTrue( db.user.TempMap.temporary)
db.sql.execute("SELECT playlistid,trackid FROM playlist_map")
db.sql.execute("SELECT playlistid,trackid FROM temp_playlist_map")
def test_playlist_map_insert(self):
track = db.make_fake_track(1, 1.234, "Test Title", "/a/b.cde")
collection = db.user.Table.find("Collection")
db.user.Map.insert(collection, track)
with self.assertRaises(sqlite3.IntegrityError):
db.user.Map.insert(collection, track)
def test_playlist_map_delete(self):
track = db.make_fake_track(1, 1.234, "Test Title", "/a/b.cde")
playlist = db.user.Table.find("Test Playlist")
db.user.Map.insert(playlist, track)
db.user.Map.delete(playlist, track)
self.assertEqual(db.user.Map.lookup_tracks(playlist), [ ])
def test_playlist_map_lookup_tracks(self):
track1 = db.make_fake_track(1, 1.234, "Test Title", "/a/b.cde")
track2 = db.make_fake_track(2, 2.345, "Test Title 2", "/a/c.def")
playlist = db.user.Table.find("Collection")
db.user.Map.insert(playlist, track1)
db.user.Map.insert(playlist, track2)
lookup_res = db.user.Map.lookup_tracks(playlist)
self.assertEqual(lookup_res, [ track1, track2 ])
def test_playlist_map_lookup_playlists(self):
track = db.make_fake_track(1, 1.234, "Test Title", "/a/b.cde")
collection = db.user.Table.find("Collection")
favorites = db.user.Table.find("Favorites")
db.user.Map.insert(collection, track)
db.user.Map.insert(favorites, track)
lookup_res = db.user.Map.lookup_playlists(track)
self.assertEqual(lookup_res, [ collection, favorites ])
def test_playlist_map_delete_track(self):
track = db.make_fake_track(1, 1.234, "Test Title", "/a/b.cde")
collection = db.user.Table.find("Collection")
favorites = db.user.Table.find("Favorites")
db.user.Map.insert(collection, track)
db.user.Map.insert(favorites, track)
db.user.Map.delete_track(track)
self.assertEqual(db.user.Map.lookup_playlists(track), [ ])
def test_playlist_map_delete_playlist(self):
track1 = db.make_fake_track(1, 1.234, "Test Title", "/a/b.cde")
track2 = db.make_fake_track(2, 2.345, "Test Title 2", "/a/c.def")
playlist = db.user.Table.find("Collection")
db.user.Map.insert(playlist, track1)
db.user.Map.insert(playlist, track2)
db.user.Map.delete_playlist(playlist)
self.assertEqual(db.user.Map.lookup_tracks(playlist), [ ])

67
db/user.py
View File

@ -93,6 +93,18 @@ class UserTable(playlist.Model):
" name TEXT UNIQUE, "
" sort TEXT, "
" FOREIGN KEY (plstateid) REFERENCES playlist_states(plstateid))")
sql.execute(f"CREATE TABLE IF NOT EXISTS playlist_map "
"(playlistid INTEGER, "
" trackid INTEGER, "
" FOREIGN KEY(playlistid) REFERENCES playlists(playlistid), "
" FOREIGN KEY(trackid) REFERENCES tracks(trackid), "
" UNIQUE(playlistid, trackid))")
sql.execute(f"CREATE TEMPORARY TABLE IF NOT EXISTS temp_playlist_map "
"(playlistid INTEGER, "
" trackid INTEGER, "
" FOREIGN KEY(playlistid) REFERENCES playlists(playlistid), "
" FOREIGN KEY(trackid) REFERENCES tracks(trackid), "
" UNIQUE(playlistid, trackid))")
sql.execute("CREATE INDEX IF NOT EXISTS playlist_index ON playlists(name)")
self.find("Collection", loop=True)
@ -101,6 +113,11 @@ class UserTable(playlist.Model):
self.find("Previous", sort=["temp_playlist_map.rowid DESC"])
self.find("Queued Tracks", sort=["playlist_map.rowid ASC"])
def do_drop(self):
sql.execute("DROP TABLE playlists")
sql.execute("DROP TABLE playlist_map")
sql.execute("DROP TABLE temp_playlist_map")
def do_factory(self, row):
if row["name"] == "Collection":
return Collection(row)
@ -127,54 +144,4 @@ class UserTable(playlist.Model):
return res
class PlaylistMap(objects.Map):
def __init__(self, temp=False):
name = "playlist_map" if temp==False else "temp_playlist_map"
self.temporary = temp
objects.Map.__init__(self, name, Table.get, self.get_track)
self.lookup_tracks = self.lookup_rhs
self.lookup_playlists = self.lookup_lhs
def get_track(self, rowid):
from . import track
return track.Table.get(rowid)
def do_create(self):
temp = "" if self.temporary == False else "TEMPORARY"
sql.execute(f"CREATE {temp} TABLE IF NOT EXISTS {self.map_name} "
"(playlistid INTEGER, "
" trackid INTEGER, "
" FOREIGN KEY(playlistid) REFERENCES playlists(playlistid), "
" FOREIGN KEY(trackid) REFERENCES tracks(trackid), "
" UNIQUE(playlistid, trackid))")
def do_insert(self, playlist, track):
sql.execute(f"INSERT INTO {self.map_name} (playlistid, trackid) "
"VALUES (?, ?)", [ playlist.rowid, track.rowid ])
def do_delete(self, playlist, track):
return sql.execute(f"DELETE FROM {self.map_name} "
"WHERE playlistid=? AND trackid=?",
[ playlist.rowid, track.rowid ])
def do_lookup_rhs(self, playlist):
return sql.execute(f"SELECT trackid FROM {self.map_name} "
"WHERE playlistid=?", [ playlist.rowid ])
def do_lookup_lhs(self, track):
return sql.execute(f"SELECT playlistid FROM {self.map_name} "
"WHERE trackid=?", [ track.rowid ])
def delete_track(self, track):
for playlist in self.lookup_playlists(track):
self.delete(playlist, track)
def delete_playlist(self, playlist):
for track in self.lookup_tracks(playlist):
self.delete(playlist, track)
Table = UserTable()
Map = PlaylistMap(temp=False)
TempMap = PlaylistMap(temp=True)

2
scanner/test_task.py
View File

@ -61,7 +61,7 @@ class TestScannerFileTask(unittest.TestCase):
self.assertEqual(genre.get_track(0), track)
new = db.user.Table.find("New Tracks")
self.assertEqual(db.user.TempMap.lookup_tracks(new), [ track ])
self.assertEqual(new.get_track(0), track)
class TestScannerImportTask(unittest.TestCase):