Porting supysonic.api.media

setup.cfg

@@ -54,7 +54,7 @@ install_requires =
     click
     flask >=0.11
     peewee
-    Pillow
+    Pillow >=9.1.0
     requests >=1.0.0
     mediafile
     watchdog >=0.8.0

supysonic/api/media.py

@@ -210,9 +210,12 @@ def stream_media():
 
     res.play_count = res.play_count + 1
     res.last_play = now()
+    res.save()
+
     user = request.user
     user.last_play = res
     user.last_play_date = now()
+    user.save()
 
     return response
 
@@ -237,16 +240,16 @@ def download_media():
         try:
             rv = Track[uid]
             return send_file(rv.path, mimetype=rv.mimetype, conditional=True)
-        except ObjectNotFound:
+        except Track.DoesNotExist:
             try:  # Album -> stream zipped tracks
                 rv = Album[uid]
-            except ObjectNotFound:
+            except Album.DoesNotExist as e:
-                raise NotFound("Track or Album")
+                raise NotFound("Track or Album") from e
     else:
         try:  # Folder -> stream zipped tracks, non recursive
             rv = Folder[fid]
-        except ObjectNotFound:
+        except Folder.DoesNotExist as e:
-            raise NotFound("Folder")
+            raise NotFound("Folder") from e
 
     # Stream a zip of multiple files to the client
     z = ZipStream(sized=True)
@@ -282,17 +285,16 @@ def download_media():
     return resp
 
 
-def _cover_from_track(tid):
+def _cover_from_track(obj):
     """Extract and return a path to a track's cover art
 
     Returns None if no cover art is available.
     """
     cache = current_app.cache
-    cache_key = "{}-cover".format(tid)
+    cache_key = "{}-cover".format(obj.id)
     try:
         return cache.get(cache_key)
     except CacheMiss:
-        obj = Track[tid]
         try:
             return cache.set(cache_key, mediafile.MediaFile(obj.path).art)
         except mediafile.UnreadableFileError:
@@ -311,14 +313,16 @@ def _cover_from_collection(obj, extract=True):
         cover_path = os.path.join(obj.path, obj.cover_art)
 
     elif isinstance(obj, Album):
-        track_with_folder_cover = obj.tracks.select(
+        track_with_folder_cover = (
-            lambda t: t.folder.cover_art is not None
+            obj.tracks.join(Folder, on=Track.folder)
-        ).first()
+            .where(Folder.cover_art.is_null(False))
+            .first()
+        )
         if track_with_folder_cover is not None:
             cover_path = _cover_from_collection(track_with_folder_cover.folder)
 
         if not cover_path and extract:
-            track_with_embedded = obj.tracks.select(lambda t: t.has_art).first()
+            track_with_embedded = obj.tracks.where(Track.has_art).first()
             if track_with_embedded is not None:
                 cover_path = _cover_from_track(track_with_embedded.id)
 
@@ -327,11 +331,7 @@ def _cover_from_collection(obj, extract=True):
     return cover_path
 
 
-@api_routing("/getCoverArt")
+def _get_cover_path(eid):
-def cover_art():
-    cache = current_app.cache
-
-    eid = request.values["id"]
     try:
         fid = get_entity_id(Folder, eid)
     except GenericError:
@@ -344,15 +344,31 @@ def cover_art():
     if not fid and not uid:
         raise GenericError("Invalid ID")
 
-    cover_path = None
+    if fid:
-    if fid and Folder.exists(id=eid):
+        try:
-        cover_path = _cover_from_collection(get_entity(Folder))
+            return _cover_from_collection(Folder[fid])
-    elif uid and Track.exists(id=eid):
+        except Folder.DoesNotExist:
-        cover_path = _cover_from_track(eid)
+            pass
-    elif uid and Album.exists(id=uid):
+    elif uid:
-        cover_path = _cover_from_collection(get_entity(Album))
+        try:
-    else:
+            return _cover_from_track(Track[uid])
-        raise NotFound("Entity")
+        except Track.DoesNotExist:
+            pass
+
+        try:
+            return _cover_from_collection(Album[uid])
+        except Album.DoesNotExist:
+            pass
+
+    raise NotFound("Entity")
+
+
+@api_routing("/getCoverArt")
+def cover_art():
+    cache = current_app.cache
+
+    eid = request.values["id"]
+    cover_path = _get_cover_path(eid)
 
     if not cover_path:
         raise NotFound("Cover art")
@@ -365,7 +381,7 @@ def cover_art():
         # extension for Flask to derive the mimetype from - derive it from the
         # contents instead.
         mimetype = None
-        if uid and os.path.splitext(cover_path)[1].lower() not in EXTENSIONS:
+        if os.path.splitext(cover_path)[1].lower() not in EXTENSIONS:
             with Image.open(cover_path) as im:
                 mimetype = "image/{}".format(im.format.lower())
         return send_file(cover_path, mimetype=mimetype)
@@ -379,7 +395,7 @@ def cover_art():
         try:
             return send_file(cache.get(cache_key), mimetype=mimetype)
         except CacheMiss:
-            im.thumbnail([size, size], Image.ANTIALIAS)
+            im.thumbnail([size, size], Image.Resampling.LANCZOS)
             with cache.set_fileobj(cache_key) as fp:
                 im.save(fp, im.format)
             return send_file(cache.get(cache_key), mimetype=mimetype)

tests/api/test_media.py

@@ -1,7 +1,7 @@
 # This file is part of Supysonic.
 # Supysonic is a Python implementation of the Subsonic server API.
 #
-# Copyright (C) 2017-2020 Alban 'spl0k' Féron
+# Copyright (C) 2017-2022 Alban 'spl0k' Féron
 #
 # Distributed under terms of the GNU AGPLv3 license.
 
@@ -12,7 +12,6 @@ import uuid
 from contextlib import closing
 from io import BytesIO
 from PIL import Image
-from pony.orm import db_session
 
 from supysonic.db import Folder, Artist, Album, Track
 
@@ -23,52 +22,51 @@ class MediaTestCase(ApiTestBase):
     def setUp(self):
         super().setUp()
 
-        with db_session:
+        folder = Folder.create(
-            folder = Folder(
+            name="Root",
-                name="Root",
+            path=os.path.abspath("tests/assets"),
-                path=os.path.abspath("tests/assets"),
+            root=True,
-                root=True,
+            cover_art="cover.jpg",
-                cover_art="cover.jpg",
+        )
-            )
+        folder = Folder.get(name="Root")
-            folder = Folder.get(name="Root")
+        self.folderid = folder.id
-            self.folderid = folder.id
 
-            artist = Artist(name="Artist")
+        artist = Artist.create(name="Artist")
-            album = Album(artist=artist, name="Album")
+        album = Album.create(artist=artist, name="Album")
 
-            track = Track(
+        track = Track.create(
-                title="23bytes",
+            title="23bytes",
+            number=1,
+            disc=1,
+            artist=artist,
+            album=album,
+            path=os.path.abspath("tests/assets/23bytes"),
+            root_folder=folder,
+            folder=folder,
+            duration=2,
+            bitrate=320,
+            last_modification=0,
+        )
+        self.trackid = track.id
+
+        self.formats = ["mp3", "flac", "ogg", "m4a"]
+        for i in range(len(self.formats)):
+            track_embeded_art = Track.create(
+                title="[silence]",
                 number=1,
                 disc=1,
                 artist=artist,
                 album=album,
-                path=os.path.abspath("tests/assets/23bytes"),
+                path=os.path.abspath(
+                    "tests/assets/formats/silence.{}".format(self.formats[i])
+                ),
                 root_folder=folder,
                 folder=folder,
                 duration=2,
                 bitrate=320,
                 last_modification=0,
             )
-            self.trackid = track.id
+            self.formats[i] = track_embeded_art.id
-
-            self.formats = ["mp3", "flac", "ogg", "m4a"]
-            for i in range(len(self.formats)):
-                track_embeded_art = Track(
-                    title="[silence]",
-                    number=1,
-                    disc=1,
-                    artist=artist,
-                    album=album,
-                    path=os.path.abspath(
-                        "tests/assets/formats/silence.{}".format(self.formats[i])
-                    ),
-                    root_folder=folder,
-                    folder=folder,
-                    duration=2,
-                    bitrate=320,
-                    last_modification=0,
-                )
-                self.formats[i] = track_embeded_art.id
 
     def test_stream(self):
         self._make_request("stream", error=10)
@@ -98,8 +96,7 @@ class MediaTestCase(ApiTestBase):
         ) as rv:
             self.assertEqual(rv.status_code, 200)
             self.assertEqual(len(rv.data), 23)
-        with db_session:
+        self.assertEqual(Track[self.trackid].play_count, 1)
-            self.assertEqual(Track[self.trackid].play_count, 1)
 
     def test_download(self):
         self._make_request("download", error=10)
@@ -120,8 +117,7 @@ class MediaTestCase(ApiTestBase):
         ) as rv:
             self.assertEqual(rv.status_code, 200)
             self.assertEqual(len(rv.data), 23)
-        with db_session:
+        self.assertEqual(Track[self.trackid].play_count, 0)
-            self.assertEqual(Track[self.trackid].play_count, 0)
 
         # dowload folder
         rv = self.client.get(