[gnome-music/wip/mschraal/type-checking: 3/3] grltrackerwrapper: Add static type checking
- From: Marinus Schraal <mschraal src gnome org>
- To: commits-list gnome org
- Cc:
- Subject: [gnome-music/wip/mschraal/type-checking: 3/3] grltrackerwrapper: Add static type checking
- Date: Tue, 6 Oct 2020 21:23:01 +0000 (UTC)
commit 92e947fdb23b34d375ea056c1dfce7df40ee6039
Author: Marinus Schraal <mschraal gnome org>
Date: Sat May 30 17:21:46 2020 +0200
grltrackerwrapper: Add static type checking
gnomemusic/grilowrappers/grltrackerwrapper.py | 232 ++++++++++++++++----------
1 file changed, 144 insertions(+), 88 deletions(-)
---
diff --git a/gnomemusic/grilowrappers/grltrackerwrapper.py b/gnomemusic/grilowrappers/grltrackerwrapper.py
index 4d30a522..bcd8bbe2 100644
--- a/gnomemusic/grilowrappers/grltrackerwrapper.py
+++ b/gnomemusic/grilowrappers/grltrackerwrapper.py
@@ -22,6 +22,9 @@
# code, but you are not obligated to do so. If you do not wish to do so,
# delete this exception statement from your version.
+from typing import Callable, Dict, List, Optional
+import typing
+
import gi
gi.require_versions({"Gfm": "0.1", "Grl": "0.3", "Tracker": "3.0"})
from gi.repository import Gfm, Gio, Grl, GLib, GObject, Tracker
@@ -30,17 +33,24 @@ from gnomemusic.corealbum import CoreAlbum
from gnomemusic.coreartist import CoreArtist
from gnomemusic.coredisc import CoreDisc
from gnomemusic.coresong import CoreSong
-from gnomemusic.grilowrappers.grltrackerplaylists import GrlTrackerPlaylists
+from gnomemusic.grilowrappers.grltrackerplaylists import (
+ GrlTrackerPlaylists, Playlist)
from gnomemusic.storeart import StoreArt
+from gnomemusic.trackerwrapper import TrackerWrapper
+if typing.TYPE_CHECKING:
+ from gnomemusic.application import Application
+ from gnomemusic.coremodel import CoreModel
+ from gnomemusic.musiclogger import MusicLogger
+ from gnomemusic.notificationmanager import NotificationManager
class GrlTrackerWrapper(GObject.GObject):
"""Wrapper for the Grilo Tracker source.
"""
- _SPLICE_SIZE = 100
+ _SPLICE_SIZE: int = 100
- _METADATA_ALBUM_CHANGED_KEYS = [
+ _METADATA_ALBUM_CHANGED_KEYS: List[int] = [
Grl.METADATA_KEY_ARTIST,
Grl.METADATA_KEY_ALBUM_ARTIST,
Grl.METADATA_KEY_COMPOSER,
@@ -50,7 +60,7 @@ class GrlTrackerWrapper(GObject.GObject):
Grl.METADATA_KEY_URL
]
- _METADATA_SONG_FILL_KEYS = [
+ _METADATA_SONG_FILL_KEYS: List[int] = [
Grl.METADATA_KEY_ALBUM,
Grl.METADATA_KEY_ALBUM_DISC_NUMBER,
Grl.METADATA_KEY_ARTIST,
@@ -63,7 +73,7 @@ class GrlTrackerWrapper(GObject.GObject):
Grl.METADATA_KEY_URL
]
- _METADATA_SONG_MEDIA_QUERY_KEYS = [
+ _METADATA_SONG_MEDIA_QUERY_KEYS: List[int] = [
Grl.METADATA_KEY_ALBUM,
Grl.METADATA_KEY_ALBUM_DISC_NUMBER,
Grl.METADATA_KEY_ARTIST,
@@ -76,12 +86,14 @@ class GrlTrackerWrapper(GObject.GObject):
Grl.METADATA_KEY_URL
]
- _METADATA_THUMBNAIL_KEYS = [
+ _METADATA_THUMBNAIL_KEYS: List[int] = [
Grl.METADATA_KEY_ID,
Grl.METADATA_KEY_THUMBNAIL
]
- def __init__(self, source, application, tracker_wrapper):
+ def __init__(
+ self, source: Grl.Source, application: "Application",
+ tracker_wrapper: TrackerWrapper) -> None:
"""Initialize the Tracker wrapper
:param Grl.TrackerSource source: The Tracker source to wrap
@@ -90,27 +102,30 @@ class GrlTrackerWrapper(GObject.GObject):
"""
super().__init__()
- self._application = application
- coremodel = application.props.coremodel
- self._log = application.props.log
+ self._application: Application = application
+ cm: CoreModel = application.props.coremodel
+ self._log: MusicLogger = application.props.log
self._songs_model = Gio.ListStore.new(CoreSong)
- coremodel.props.songs_proxy.append(self._songs_model)
- self._source = None
+ cm.props.songs_proxy.append(self._songs_model)
+ self._source: Optional[Grl.Source] = None
self._albums_model = Gio.ListStore.new(CoreAlbum)
- coremodel.props.albums_proxy.append(self._albums_model)
- self._album_ids = {}
+ cm.props.albums_proxy.append(self._albums_model)
+ self._album_ids: Dict[int, CoreAlbum] = {}
self._artists_model = Gio.ListStore.new(CoreArtist)
- coremodel.props.artists_proxy.append(self._artists_model)
- self._artist_ids = {}
- self._hash = {}
- self._song_search_proxy = coremodel.props.songs_search_proxy
- self._album_search_model = coremodel.props.albums_search
- self._artist_search_model = coremodel.props.artists_search
- self._batch_changed_media_ids = {}
- self._content_changed_timeout = 0
- self._tracker_playlists = None
- self._tracker_wrapper = tracker_wrapper
- self._notificationmanager = application.props.notificationmanager
+ cm.props.artists_proxy.append(self._artists_model)
+ self._artist_ids: Dict[int, CoreArtist] = {}
+ self._hash: Dict[int, CoreSong] = {}
+ self._song_search_proxy: Gio.ListStore = cm.props.songs_search_proxy
+ self._album_search_model: Gfm.FilterListModel = cm.props.albums_search
+ self._artist_search_model: Gfm.FilterListModel = (
+ cm.props.artists_search)
+ self._batch_changed_media_ids: Dict[
+ Grl.SourceChangeType, List[int]] = {}
+ self._content_changed_timeout: int = 0
+ self._tracker_playlists: Optional[GrlTrackerPlaylists] = None
+ self._tracker_wrapper: TrackerWrapper = tracker_wrapper
+ self._notificationmanager: NotificationManager = (
+ application.props.notificationmanager)
self._song_search_tracker = Gfm.FilterListModel.new(self._songs_model)
self._song_search_tracker.set_filter_func(lambda a: False)
@@ -131,11 +146,11 @@ class GrlTrackerWrapper(GObject.GObject):
self._initial_artists_fill()
@GObject.Property(type=Grl.Source, default=None)
- def source(self):
+ def source(self) -> Grl.Source:
return self._source
@source.setter # type: ignore
- def source(self, new_source):
+ def source(self, new_source: Grl.Source):
"""Set a new grilo tracker source
Everytime, the tracker plugin is loaded, a new source is
@@ -152,21 +167,25 @@ class GrlTrackerWrapper(GObject.GObject):
self._content_changed_id = self._source.connect(
"content-changed", self._batch_content_changed)
- def _batch_content_changed(self, source, medias, change_type, loc_unknown):
+ def _batch_content_changed(
+ self, source: Grl.Source, medias: List[Grl.Media],
+ change_type: Grl.SourceChangeType, loc_unknown: bool) -> None:
if medias == []:
return
if change_type not in self._batch_changed_media_ids.keys():
self._batch_changed_media_ids[change_type] = []
- [self._batch_changed_media_ids[change_type].append(media.get_id())
- for media in medias if media.is_audio() or media.is_container()]
+ [self._batch_changed_media_ids[ # type: ignore
+ change_type].append(media.get_id())
+ for media in medias
+ if media.is_audio() or media.is_container()]
if self._content_changed_timeout == 0:
self._content_changed_timeout = GLib.timeout_add(
250, self._on_content_changed)
- def _on_content_changed(self):
+ def _on_content_changed(self) -> bool:
for change_type in self._batch_changed_media_ids.keys():
media_ids = self._batch_changed_media_ids[change_type]
@@ -193,8 +212,8 @@ class GrlTrackerWrapper(GObject.GObject):
return GLib.SOURCE_REMOVE
- def _check_album_change(self):
- album_ids = {}
+ def _check_album_change(self) -> None:
+ album_ids: Dict[int, CoreAlbum] = {}
query = """
SELECT
@@ -232,7 +251,9 @@ class GrlTrackerWrapper(GObject.GObject):
'location_filter': self._tracker_wrapper.location_filter()
}
- def check_album_cb(source, op_id, media, remaining, error):
+ def check_album_cb(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: str) -> None:
if error:
self._log.warning("Error: {}".format(error))
return
@@ -267,8 +288,8 @@ class GrlTrackerWrapper(GObject.GObject):
query, self._METADATA_ALBUM_CHANGED_KEYS, self._fast_options,
check_album_cb)
- def _check_artist_change(self):
- artist_ids = {}
+ def _check_artist_change(self) -> None:
+ artist_ids: Dict[int, CoreArtist] = {}
query = """
SELECT ?type ?id ?artist
@@ -307,7 +328,9 @@ class GrlTrackerWrapper(GObject.GObject):
Grl.METADATA_KEY_ID
]
- def check_artist_cb(source, op_id, media, remaining, error):
+ def check_artist_cb(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
return
@@ -340,7 +363,7 @@ class GrlTrackerWrapper(GObject.GObject):
self.props.source.query(
query, metadata_keys, self._fast_options, check_artist_cb)
- def _remove_media(self, media_ids):
+ def _remove_media(self, media_ids: List[int]) -> None:
for media_id in media_ids:
try:
coresong = self._hash.pop(media_id)
@@ -356,8 +379,8 @@ class GrlTrackerWrapper(GObject.GObject):
self._songs_model.remove(idx)
break
- def _song_media_query(self, media_ids):
- media_ids = str(media_ids)[1:-1]
+ def _song_media_query(self, ids: List[int]) -> str:
+ media_ids = str(ids)[1:-1]
query = """
SELECT
@@ -403,9 +426,11 @@ class GrlTrackerWrapper(GObject.GObject):
return query
- def _changed_media(self, media_ids):
+ def _changed_media(self, media_ids: List[int]) -> None:
- def _update_changed_media(source, op_id, media, remaining, error):
+ def _update_changed_media(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
return
@@ -431,11 +456,13 @@ class GrlTrackerWrapper(GObject.GObject):
self._METADATA_SONG_MEDIA_QUERY_KEYS, self._fast_options,
_update_changed_media)
- def _initial_songs_fill(self):
+ def _initial_songs_fill(self) -> None:
self._notificationmanager.push_loading()
- songs_added = []
+ songs_added: List[int] = []
- def _add_to_model(source, op_id, media, remaining, error):
+ def _add_to_model(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
self._notificationmanager.pop_loading()
@@ -508,11 +535,13 @@ class GrlTrackerWrapper(GObject.GObject):
query, self._METADATA_SONG_FILL_KEYS, self._fast_options,
_add_to_model)
- def _initial_albums_fill(self):
+ def _initial_albums_fill(self) -> None:
self._notificationmanager.push_loading()
- albums_added = []
+ albums_added: List[int] = []
- def _add_to_albums_model(source, op_id, media, remaining, error):
+ def _add_to_albums_model(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
self._notificationmanager.pop_loading()
@@ -576,11 +605,13 @@ class GrlTrackerWrapper(GObject.GObject):
query, self._METADATA_ALBUM_CHANGED_KEYS, self._fast_options,
_add_to_albums_model)
- def _initial_artists_fill(self):
+ def _initial_artists_fill(self) -> None:
self._notificationmanager.push_loading()
- artists_added = []
+ artists_added: List[int] = []
- def _add_to_artists_model(source, op_id, media, remaining, error):
+ def _add_to_artists_model(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
self._notificationmanager.pop_loading()
@@ -632,7 +663,7 @@ class GrlTrackerWrapper(GObject.GObject):
'location_filter': self._tracker_wrapper.location_filter()
}
- metadata_keys = [
+ metadata_keys: List[int] = [
Grl.METADATA_KEY_ARTIST,
Grl.METADATA_KEY_ID
]
@@ -640,7 +671,8 @@ class GrlTrackerWrapper(GObject.GObject):
self.props.source.query(
query, metadata_keys, self._fast_options, _add_to_artists_model)
- def get_artist_albums(self, media, model):
+ def get_artist_albums(
+ self, media: Grl.Source, model: Gfm.FilterListModel) -> None:
"""Get all albums by an artist
:param Grl.Media media: The media with the artist id
@@ -682,9 +714,11 @@ class GrlTrackerWrapper(GObject.GObject):
'location_filter': self._tracker_wrapper.location_filter()
}
- albums = []
+ albums: List[Grl.Media] = []
- def query_cb(source, op_id, media, remaining, error):
+ def query_cb(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
self._notificationmanager.pop_loading()
@@ -697,7 +731,8 @@ class GrlTrackerWrapper(GObject.GObject):
albums.append(media)
- def albums_filter(corealbum, albums):
+ def albums_filter(
+ corealbum: CoreAlbum, albums: List[Grl.Media]) -> bool:
for media in albums:
if media.get_id() == corealbum.props.media.get_id():
return True
@@ -707,7 +742,8 @@ class GrlTrackerWrapper(GObject.GObject):
self.props.source.query(
query, [Grl.METADATA_KEY_TITLE], self._fast_options, query_cb)
- def get_album_discs(self, media, disc_model):
+ def get_album_discs(
+ self, media: Grl.Media, disc_model: Gfm.SortListModel) -> None:
"""Get all discs of an album
:param Grl.Media media: The media with the album id
@@ -744,7 +780,9 @@ class GrlTrackerWrapper(GObject.GObject):
'location_filter': self._tracker_wrapper.location_filter()
}
- def _disc_nr_cb(source, op_id, media, remaining, error):
+ def _disc_nr_cb(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
self._notificationmanager.pop_loading()
@@ -762,7 +800,8 @@ class GrlTrackerWrapper(GObject.GObject):
query, [Grl.METADATA_KEY_ALBUM_DISC_NUMBER], self._fast_options,
_disc_nr_cb)
- def populate_album_disc_songs(self, media, disc_nr, callback):
+ def populate_album_disc_songs(
+ self, media: Grl.Media, disc_nr: int, callback: Callable) -> None:
# FIXME: Pass a model and fill it.
# FIXME: The query is similar to the other song queries, reuse
# if possible.
@@ -772,7 +811,7 @@ class GrlTrackerWrapper(GObject.GObject):
:param int disc_nr: The disc number
:param callback: The callback to call for every song added
"""
- album_id = media.get_id()
+ album_id: str = media.get_id()
query = """
SELECT
@@ -824,7 +863,7 @@ class GrlTrackerWrapper(GObject.GObject):
'miner_fs_busname': self._tracker_wrapper.props.miner_fs_busname
}
- metadata_keys = [
+ metadata_keys: List[int] = [
Grl.METADATA_KEY_ALBUM,
Grl.METADATA_KEY_ALBUM_DISC_NUMBER,
Grl.METADATA_KEY_ARTIST,
@@ -839,13 +878,13 @@ class GrlTrackerWrapper(GObject.GObject):
self.props.source.query(
query, metadata_keys, self._fast_options, callback)
- def search(self, text):
+ def search(self, text: str) -> None:
# FIXME: Searches are limited to not bog down the UI with
# widget creation ({List,Flow}Box limitations). The limit is
# arbitrarily set to 50 and set in the Tracker query. It should
# be possible to set it through Grilo options instead. This
# does not work as expected and needs further investigation.
- term = Tracker.sparql_escape_string(
+ term: str = Tracker.sparql_escape_string(
GLib.utf8_normalize(
GLib.utf8_casefold(text, -1), -1, GLib.NormalizeMode.NFKD))
@@ -903,12 +942,14 @@ class GrlTrackerWrapper(GObject.GObject):
'name': term
}
- artist_filter_ids = []
+ artist_filter_ids: List[str] = []
- def artist_filter(coreartist):
+ def artist_filter(coreartist: CoreArtist) -> bool:
return coreartist.media.get_id() in artist_filter_ids
- def artist_search_cb(source, op_id, media, remaining, error):
+ def artist_search_cb(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
self._notificationmanager.pop_loading()
@@ -970,12 +1011,14 @@ class GrlTrackerWrapper(GObject.GObject):
'name': term
}
- album_filter_ids = []
+ album_filter_ids: List[str] = []
- def album_filter(corealbum):
+ def album_filter(corealbum: CoreAlbum) -> bool:
return corealbum.media.get_id() in album_filter_ids
- def albums_search_cb(source, op_id, media, remaining, error):
+ def albums_search_cb(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
self._notificationmanager.pop_loading()
@@ -1042,12 +1085,14 @@ class GrlTrackerWrapper(GObject.GObject):
'name': term
}
- filter_ids = []
+ filter_ids: List[str] = []
- def songs_filter(coresong):
+ def songs_filter(coresong: CoreSong) -> bool:
return coresong.media.get_id() in filter_ids
- def songs_search_cb(source, op_id, media, remaining, error):
+ def songs_search_cb(
+ source: Grl.Source, op_id: int, media: Grl.Media,
+ remaining: int, error: Optional[GLib.Error]) -> None:
if error:
self._log.warning("Error: {}".format(error))
self._notificationmanager.pop_loading()
@@ -1063,7 +1108,7 @@ class GrlTrackerWrapper(GObject.GObject):
self.props.source.query(
query, [Grl.METADATA_KEY_ID], self._fast_options, songs_search_cb)
- def _get_album_for_media_id_query(self, media_id, song=True):
+ def _get_album_for_media_id_query(self, media_id: str, song=True) -> str:
# Even though we check for the album_artist, we fill
# the artist key, since Grilo coverart plugins use
# only that key for retrieval.
@@ -1123,7 +1168,7 @@ class GrlTrackerWrapper(GObject.GObject):
return query
- def get_song_art(self, coresong):
+ def get_song_art(self, coresong: CoreSong) -> None:
"""Retrieve song art for the given CoreSong
Since MediaArt does not really support per-song art this
@@ -1132,7 +1177,7 @@ class GrlTrackerWrapper(GObject.GObject):
:param CoreSong coresong: CoreSong to get art for
"""
- media = coresong.props.media
+ media: Grl.Media = coresong.props.media
# If there is no album and artist do not go through with the
# query, it will not give any results.
@@ -1159,19 +1204,19 @@ class GrlTrackerWrapper(GObject.GObject):
else:
StoreArt(coresong, thumbnail_uri)
- song_id = media.get_id()
- query = self._get_album_for_media_id_query(song_id)
+ song_id: str = media.get_id()
+ query: str = self._get_album_for_media_id_query(song_id)
self.props.source.query(
query, self._METADATA_THUMBNAIL_KEYS, self._full_options,
art_retrieved_cb)
- def get_album_art(self, corealbum):
+ def get_album_art(self, corealbum: CoreAlbum) -> None:
"""Retrieve album art for the given CoreAlbum
:param CoreAlbum corealbum: CoreAlbum to get art for
"""
- media = corealbum.props.media
+ media: Grl.Media = corealbum.props.media
def art_retrieved_cb(source, op_id, queried_media, remaining, error):
if error:
@@ -1185,27 +1230,27 @@ class GrlTrackerWrapper(GObject.GObject):
# empty result can be ignored.
return
- thumbnail_uri = queried_media.get_thumbnail()
+ thumbnail_uri: str = queried_media.get_thumbnail()
if thumbnail_uri is None:
corealbum.props.thumbnail = "generic"
else:
StoreArt(corealbum, thumbnail_uri)
- album_id = media.get_id()
- query = self._get_album_for_media_id_query(album_id, False)
+ album_id: str = media.get_id()
+ query: str = self._get_album_for_media_id_query(album_id, False)
- self._source.query(
+ self.props.source.query(
query, self._METADATA_THUMBNAIL_KEYS, self._full_options,
art_retrieved_cb)
- def get_artist_art(self, coreartist):
+ def get_artist_art(self, coreartist: CoreArtist) -> None:
"""Retrieve artist art for the given CoreArtist
This retrieves art through Grilo online services only.
:param CoreArtist coreartist: CoreArtist to get art for
"""
- media = coreartist.props.media
+ media: Grl.Media = coreartist.props.media
def art_resolved_cb(source, op_id, resolved_media, data, error):
if error:
@@ -1223,25 +1268,36 @@ class GrlTrackerWrapper(GObject.GObject):
media, [Grl.METADATA_KEY_THUMBNAIL], self._full_options,
art_resolved_cb, None)
- def stage_playlist_deletion(self, playlist):
+ def stage_playlist_deletion(self, playlist: Optional[Playlist]) -> None:
"""Prepares playlist deletion.
:param Playlist playlist: playlist
"""
+ if self._tracker_playlists is None:
+ return
+
self._tracker_playlists.stage_playlist_deletion(playlist)
- def finish_playlist_deletion(self, playlist, deleted):
+ def finish_playlist_deletion(
+ self, playlist: Optional[Playlist], deleted: bool) -> None:
"""Finishes playlist deletion.
:param Playlist playlist: playlist
:param bool deleted: indicates if the playlist has been deleted
"""
+ if self._tracker_playlists is None:
+ return
+
self._tracker_playlists.finish_playlist_deletion(playlist, deleted)
- def create_playlist(self, playlist_title, callback):
+ def create_playlist(
+ self, playlist_title: str, callback: Callable) -> None:
"""Creates a new user playlist.
:param str playlist_title: playlist title
:param callback: function to perform once, the playlist is created
"""
+ if self._tracker_playlists is None:
+ return
+
self._tracker_playlists.create_playlist(playlist_title, callback)
[
Date Prev][
Date Next] [
Thread Prev][
Thread Next]
[
Thread Index]
[
Date Index]
[
Author Index]