From 2f2e0a0fed925ecec4b34c295fcdda9d6d8a036e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0ar=C5=ABnas=20Nejus?= Date: Mon, 17 Aug 2026 21:23:58 +0100 Subject: [PATCH 1/2] typing: add annotations to API and export plugins --- beetsplug/aura.py | 89 +++++++++++++++++++++---------------- beetsplug/export.py | 46 ++++++++++++------- beetsplug/web/__init__.py | 93 ++++++++++++++++++++++----------------- 3 files changed, 134 insertions(+), 94 deletions(-) diff --git a/beetsplug/aura.py b/beetsplug/aura.py index 912c7a8628..2f469731be 100644 --- a/beetsplug/aura.py +++ b/beetsplug/aura.py @@ -6,7 +6,7 @@ import re from dataclasses import dataclass from mimetypes import guess_type -from typing import TYPE_CHECKING, ClassVar, Protocol +from typing import TYPE_CHECKING, Any, ClassVar, Protocol from flask import ( Blueprint, @@ -27,11 +27,14 @@ from beets.ui import Subcommand, _open_library if TYPE_CHECKING: - from collections.abc import Mapping + from collections.abc import Iterable, Mapping, Sequence from beets.dbcore.query import SQLiteType + from beets.dbcore.sort import Sort from beets.library import LibModel, Library + from ._typing import JSONDict + class AuraCLIOpts(Protocol): debug: bool @@ -127,7 +130,7 @@ def from_app(cls) -> Self: return cls(current_app.config["lib"], request.args) @staticmethod - def error(status, title, detail): + def error(status: str, title: str, detail: str) -> Any: """Make a response for an error following the JSON:API spec. Args: @@ -157,7 +160,7 @@ def get_attribute_converter(cls, beets_attr: str) -> type[SQLiteType]: # Fall back to string (NOTE: probably not good) return str - def translate_filters(self): + def translate_filters(self) -> AndQuery: """Translate filters from request arguments to a beets Query.""" # The format of each filter key in the request parameter is: # filter[]. This regex extracts . @@ -180,7 +183,7 @@ def translate_filters(self): # NOTE: AURA doesn't officially support multiple queries return AndQuery(queries) - def translate_sorts(self, sort_arg): + def translate_sorts(self, sort_arg: str) -> MultipleSort: """Translate an AURA sort parameter into a beets Sort. Args: @@ -205,7 +208,9 @@ def translate_sorts(self, sort_arg): sorts.append(SlowFieldSort(beets_attr, ascending=ascending)) return MultipleSort(sorts) - def paginate(self, collection): + def paginate( + self, collection: Sequence[Any] + ) -> tuple[list[Any], str | None]: """Get a page of the collection and the URL to the next page. Args: @@ -245,7 +250,9 @@ def paginate(self, collection): ] return data, next_url - def get_included(self, data, include_str): + def get_included( + self, data: Iterable[JSONDict], include_str: str + ) -> list[JSONDict | None]: """Build a list of resource objects for inclusion. Args: @@ -271,7 +278,7 @@ def get_included(self, data, include_str): if identifier not in unique_identifiers: unique_identifiers.append(identifier) # TODO: I think this could be improved - included = [] + included: list[JSONDict | None] = [] for identifier in unique_identifiers: res_type = identifier["type"] if res_type == "track": @@ -300,7 +307,7 @@ def get_included(self, data, include_str): raise ValueError(f"Invalid resource type: {res_type}") return included - def all_resources(self): + def all_resources(self) -> JSONDict: """Build document for /tracks, /albums or /artists.""" query = self.translate_filters() sort_arg = self.args.get("sort", None) @@ -320,7 +327,7 @@ def all_resources(self): collection = self.get_collection(query=query, sort=sort) # Convert info to AURA form and paginate it data, next_url = self.paginate(collection) - document = {"data": data} + document: JSONDict = {"data": data} # If there are more pages then provide a way to access them if next_url: document["links"] = {"next": next_url} @@ -330,14 +337,14 @@ def all_resources(self): document["included"] = self.get_included(data, include_str) return document - def single_resource_document(self, resource_object): + def single_resource_document(self, resource_object: JSONDict) -> JSONDict: """Build document for a specific requested resource. Args: resource_object: A dictionary in the form of a JSON:API resource object. """ - document = {"data": resource_object} + document: JSONDict = {"data": resource_object} include_str = self.args.get("include", None) if include_str: # [document["data"]] is because arg needs to be list @@ -354,7 +361,9 @@ class TrackDocument(AURADocument): attribute_map = TRACK_ATTR_MAP - def get_collection(self, query=None, sort=None): + def get_collection( + self, query: list[str] | None = None, sort: Sort | None = None + ) -> Any: """Get Item objects from the library. Args: @@ -377,7 +386,7 @@ def get_attribute_converter(cls, beets_attr: str) -> type[SQLiteType]: return super().get_attribute_converter(beets_attr) @staticmethod - def get_resource_object(lib: Library, track): + def get_resource_object(lib: Library, track: Item) -> JSONDict: """Construct a JSON:API resource object from a beets Item. Args: @@ -409,7 +418,7 @@ def get_resource_object(lib: Library, track): "relationships": relationships, } - def single_resource(self, track_id): + def single_resource(self, track_id: int) -> Any: """Get track from the library and build a document. Args: @@ -434,7 +443,9 @@ class AlbumDocument(AURADocument): attribute_map = ALBUM_ATTR_MAP - def get_collection(self, query=None, sort=None): + def get_collection( + self, query: list[str] | None = None, sort: Sort | None = None + ) -> Any: """Get Album objects from the library. Args: @@ -444,7 +455,7 @@ def get_collection(self, query=None, sort=None): return self.lib.albums(query, sort) @staticmethod - def get_resource_object(lib: Library, album): + def get_resource_object(lib: Library, album: Album) -> JSONDict: """Construct a JSON:API resource object from a beets Album. Args: @@ -493,7 +504,7 @@ def get_resource_object(lib: Library, album): "relationships": relationships, } - def single_resource(self, album_id): + def single_resource(self, album_id: int) -> Any: """Get album from the library and build a document. Args: @@ -518,7 +529,9 @@ class ArtistDocument(AURADocument): attribute_map = ARTIST_ATTR_MAP - def get_collection(self, query=None, sort=None): + def get_collection( + self, query: list[str] | None = None, sort: Sort | None = None + ) -> Any: """Get a list of artist names from the library. Args: @@ -535,7 +548,7 @@ def get_collection(self, query=None, sort=None): return collection @staticmethod - def get_resource_object(lib: Library, artist_id): + def get_resource_object(lib: Library, artist_id: str) -> JSONDict | None: """Construct a JSON:API resource object for the given artist. Args: @@ -578,7 +591,7 @@ def get_resource_object(lib: Library, artist_id): "relationships": relationships, } - def single_resource(self, artist_id): + def single_resource(self, artist_id: str) -> Any: """Get info for the requested artist and build a document. Args: @@ -594,7 +607,7 @@ def single_resource(self, artist_id): return self.single_resource_document(artist_resource) -def safe_filename(fn): +def safe_filename(fn: str) -> bool: """Check whether a string is a simple (non-path) filename. For example, `foo.txt` is safe because it is a "plain" filename. But @@ -618,7 +631,7 @@ class ImageDocument(AURADocument): model_cls = Album @staticmethod - def get_image_path(lib: Library, image_id): + def get_image_path(lib: Library, image_id: str) -> str | None: """Works out the full path to the image with the given id. Returns None if there is no such image. @@ -659,7 +672,7 @@ def get_image_path(lib: Library, image_id): return None @staticmethod - def get_resource_object(lib: Library, image_id): + def get_resource_object(lib: Library, image_id: str) -> JSONDict | None: """Construct a JSON:API resource object for the given image. Args: @@ -701,7 +714,7 @@ def get_resource_object(lib: Library, image_id): "relationships": relationships, } - def single_resource(self, image_id): + def single_resource(self, image_id: str) -> Any: """Get info for the requested image and build a document. Args: @@ -723,7 +736,7 @@ def single_resource(self, image_id): @aura_bp.route("/server") -def server_info(): +def server_info() -> dict[str, JSONDict]: """Respond with info about the server.""" return {"data": {"type": "server", "id": "0", "attributes": SERVER_INFO}} @@ -732,13 +745,13 @@ def server_info(): @aura_bp.route("/tracks") -def all_tracks(): +def all_tracks() -> Any: """Respond with a list of all tracks and related information.""" return TrackDocument.from_app().all_resources() @aura_bp.route("/tracks/") -def single_track(track_id): +def single_track(track_id: int) -> Any: """Respond with info about the specified track. Args: @@ -748,7 +761,7 @@ def single_track(track_id): @aura_bp.route("/tracks//audio") -def audio_file(track_id): +def audio_file(track_id: int) -> Any: """Supply an audio file for the specified track. Args: @@ -807,13 +820,13 @@ def audio_file(track_id): @aura_bp.route("/albums") -def all_albums(): +def all_albums() -> Any: """Respond with a list of all albums and related information.""" return AlbumDocument.from_app().all_resources() @aura_bp.route("/albums/") -def single_album(album_id): +def single_album(album_id: int) -> Any: """Respond with info about the specified album. Args: @@ -827,14 +840,14 @@ def single_album(album_id): @aura_bp.route("/artists") -def all_artists(): +def all_artists() -> Any: """Respond with a list of all artists and related information.""" return ArtistDocument.from_app().all_resources() # Using the path converter allows slashes in artist_id @aura_bp.route("/artists/") -def single_artist(artist_id): +def single_artist(artist_id: str) -> Any: """Respond with info about the specified artist. Args: @@ -850,7 +863,7 @@ def single_artist(artist_id): @aura_bp.route("/images/") -def single_image(image_id): +def single_image(image_id: str) -> Any: """Respond with info about the specified image. Args: @@ -861,7 +874,7 @@ def single_image(image_id): @aura_bp.route("/images//file") -def image_file(image_id): +def image_file(image_id: str) -> Any: """Supply an image file for the specified image. Args: @@ -881,7 +894,7 @@ def image_file(image_id): # WSGI app -def create_app(): +def create_app() -> Flask: """An application factory for use by a WSGI server.""" config["aura"].add( { @@ -929,11 +942,11 @@ def create_app(): class AURAPlugin(BeetsPlugin): """The BeetsPlugin subclass for the AURA server plugin.""" - def __init__(self): + def __init__(self) -> None: """Add configuration options for the AURA plugin.""" super().__init__() - def commands(self): + def commands(self) -> list[Subcommand]: """Add subcommand used to run the AURA server.""" def run_aura(lib: Library, opts: AuraCLIOpts, args: list[str]) -> None: diff --git a/beetsplug/export.py b/beetsplug/export.py index 4e6250ca94..52660f55a3 100644 --- a/beetsplug/export.py +++ b/beetsplug/export.py @@ -7,7 +7,7 @@ import json import sys from datetime import date, datetime -from typing import TYPE_CHECKING, Literal, Protocol, get_args +from typing import TYPE_CHECKING, Any, Literal, Protocol, get_args from xml.etree import ElementTree import mediafile @@ -17,8 +17,12 @@ from beetsplug.info import library_data, tag_data if TYPE_CHECKING: + from collections.abc import Sequence + from beets.library import Library + from ._typing import JSONDict + Format = Literal["json", "jsonlines", "csv", "xml"] @@ -26,7 +30,7 @@ class ExportCLIOpts(Protocol): library: bool | None album: bool append: bool - included_keys: list[str] + included_keys: Sequence[str] output: str | None format: Format | None @@ -34,14 +38,14 @@ class ExportCLIOpts(Protocol): class ExportEncoder(json.JSONEncoder): """Deals with dates because JSON doesn't have a standard""" - def default(self, o): + def default(self, o: object) -> Any: if isinstance(o, (datetime, date)): return o.isoformat() return json.JSONEncoder.default(self, o) class ExportPlugin(BeetsPlugin): - def __init__(self): + def __init__(self) -> None: super().__init__() self.config.add( @@ -82,7 +86,7 @@ def __init__(self): } ) - def commands(self): + def commands(self) -> list[ui.Subcommand]: cmd = ui.Subcommand("export", help="export data from beets") cmd.func = self.run cmd.parser.add_album_option() @@ -119,7 +123,9 @@ def commands(self): ) return [cmd] - def run(self, lib: Library, opts: ExportCLIOpts, args: list[str]) -> None: + def run( + self, lib: Library, opts: ExportCLIOpts, args: Sequence[str] + ) -> None: file_path = opts.output file_mode = "a" if opts.append else "w" default_format: Format = self.config["default_format"].as_choice( @@ -166,7 +172,9 @@ def run(self, lib: Library, opts: ExportCLIOpts, args: list[str]) -> None: class ExportFormat: """The output format type""" - def __init__(self, file_path, file_mode="w", encoding="utf-8"): + def __init__( + self, file_path: str, file_mode: str = "w", encoding: str = "utf-8" + ) -> None: self.path = file_path self.mode = file_mode self.encoding = encoding @@ -178,7 +186,9 @@ def __init__(self, file_path, file_mode="w", encoding="utf-8"): ) @classmethod - def factory(cls, file_type, **kwargs): + def factory( + cls, file_type: str, **kwargs + ) -> JsonFormat | CSVFormat | XMLFormat: if file_type in ["json", "jsonlines"]: return JsonFormat(**kwargs) if file_type == "csv": @@ -187,17 +197,19 @@ def factory(cls, file_type, **kwargs): return XMLFormat(**kwargs) raise NotImplementedError() - def export(self, data, **kwargs): + def export(self, data: list[JSONDict], **kwargs) -> None: raise NotImplementedError() class JsonFormat(ExportFormat): """Saves in a json file""" - def __init__(self, file_path, file_mode="w", encoding="utf-8"): + def __init__( + self, file_path: str, file_mode: str = "w", encoding: str = "utf-8" + ) -> None: super().__init__(file_path, file_mode, encoding) - def export(self, data, **kwargs): + def export(self, data: list[JSONDict], **kwargs) -> None: json.dump(data, self.out_stream, cls=ExportEncoder, **kwargs) self.out_stream.write("\n") @@ -205,10 +217,12 @@ def export(self, data, **kwargs): class CSVFormat(ExportFormat): """Saves in a csv file""" - def __init__(self, file_path, file_mode="w", encoding="utf-8"): + def __init__( + self, file_path: str, file_mode: str = "w", encoding: str = "utf-8" + ) -> None: super().__init__(file_path, file_mode, encoding) - def export(self, data, **kwargs): + def export(self, data: list[JSONDict], **kwargs) -> None: header = list(data[0].keys()) if data else [] writer = csv.DictWriter(self.out_stream, fieldnames=header, **kwargs) writer.writeheader() @@ -218,10 +232,12 @@ def export(self, data, **kwargs): class XMLFormat(ExportFormat): """Saves in a xml file""" - def __init__(self, file_path, file_mode="w", encoding="utf-8"): + def __init__( + self, file_path: str, file_mode: str = "w", encoding: str = "utf-8" + ) -> None: super().__init__(file_path, file_mode, encoding) - def export(self, data, **kwargs): + def export(self, data: list[JSONDict], **kwargs) -> None: # Creates the XML file structure. library = ElementTree.Element("library") tracks = ElementTree.SubElement(library, "tracks") diff --git a/beetsplug/web/__init__.py b/beetsplug/web/__init__.py index 86f6c93353..3dc75a7371 100644 --- a/beetsplug/web/__init__.py +++ b/beetsplug/web/__init__.py @@ -6,7 +6,7 @@ import json import os import typing as t -from typing import TYPE_CHECKING, Protocol +from typing import TYPE_CHECKING, Any, Protocol import flask from flask import jsonify @@ -19,7 +19,10 @@ from beets.plugins import BeetsPlugin if TYPE_CHECKING: - from beets.library import Library + from collections.abc import Iterable, Iterator, Sequence + + from beets.library import LibModel, Library + from beetsplug._typing import JSONDict class WebCLIOpts(Protocol): @@ -40,7 +43,7 @@ class LibraryCtx(flask.ctx._AppCtxGlobals): # Utilities. -def _rep(obj, expand=False): +def _rep(obj: LibModel, expand: bool = False) -> JSONDict | None: """Get a flat -- i.e., JSON-ish -- representation of a beets Item or Album object. For Albums, `expand` dictates whether tracks are included. @@ -78,11 +81,13 @@ def _rep(obj, expand=False): return None -def json_generator(items, root, expand=False): +def json_generator( + items: Sequence[LibModel], root: str, expand: bool = False +) -> Iterator[Any]: """Generator that dumps list of beets Items or Albums as JSON :param root: root key for JSON - :param items: list of :class:`Item` or :class:`Album` to dump + :param items: sequence of :class:`Item` or :class:`Album` to dump :param expand: If true every :class:`Album` contains its items in the json representation :returns: generator that yields strings @@ -98,13 +103,13 @@ def json_generator(items, root, expand=False): yield "]}" -def is_expand(): +def is_expand() -> bool: """Returns whether the current request is for an expanded response.""" return flask.request.args.get("expand") is not None -def is_delete(): +def is_delete() -> bool: """Returns whether the current delete request should remove the selected files. """ @@ -112,16 +117,16 @@ def is_delete(): return flask.request.args.get("delete") is not None -def get_method(): +def get_method() -> str: """Returns the HTTP method of the current request.""" return flask.request.method -def resource(name, patchable=False): +def resource(name: str, patchable: bool = False) -> Any: """Decorates a function to handle RESTful HTTP requests for a resource.""" - def make_responder(retriever): - def responder(ids): + def make_responder(retriever: t.Callable[[int], LibModel | None]) -> Any: + def responder(ids: Sequence[int]): entities = [retriever(id_) for id_ in ids] entities = [entity for entity in entities if entity] @@ -169,11 +174,13 @@ def responder(ids): return make_responder -def resource_query(name, patchable=False): +def resource_query(name: str, patchable: bool = False) -> Any: """Decorates a function to handle RESTful HTTP queries for resources.""" - def make_responder(query_func): - def responder(queries): + def make_responder( + query_func: t.Callable[[Iterable[str]], Sequence[LibModel]], + ) -> Any: + def responder(queries: Iterable[str]) -> Any: entities = query_func(queries) if get_method() == "DELETE": @@ -215,13 +222,13 @@ def responder(queries): return make_responder -def resource_list(name): +def resource_list(name: str) -> Any: """Decorates a function to handle RESTful HTTP request for a list of resources. """ - def make_responder(list_all): - def responder(): + def make_responder(list_all: t.Callable[[], Sequence[LibModel]]) -> Any: + def responder() -> Any: return app.response_class( json_generator(list_all(), root=name, expand=is_expand()), mimetype="application/json", @@ -233,7 +240,9 @@ def responder(): return make_responder -def _get_unique_table_field_values(model, field, sort_field): +def _get_unique_table_field_values( + model: type[LibModel], field: str, sort_field: str +) -> list[Any]: """retrieve all unique values belonging to a key from a model""" if field not in model.all_keys() or sort_field not in model.all_keys(): raise KeyError @@ -247,7 +256,7 @@ def _get_unique_table_field_values(model, field, sort_field): class IdListConverter(BaseConverter): """Converts comma separated lists of ids in urls to integer lists.""" - def to_python(self, value): + def to_python(self, value: str) -> list[int]: ids = [] for id_ in value.split(","): try: @@ -256,14 +265,14 @@ def to_python(self, value): pass return ids - def to_url(self, value): + def to_url(self, value: Sequence[int]) -> str: return ",".join(str(v) for v in value) class QueryConverter(PathConverter): """Converts slash separated lists of queries in the url to string list.""" - def to_python(self, value): + def to_python(self, value: str) -> list[str]: queries = value.split("/") """Do not do path substitution on regex value tests""" return [ @@ -271,7 +280,7 @@ def to_python(self, value): for query in queries ] - def to_url(self, value): + def to_url(self, value: Sequence[str]) -> str: return "/".join([v.replace(os.sep, "\\") for v in value]) @@ -289,7 +298,7 @@ class EverythingConverter(PathConverter): @app.before_request -def before_request(): +def before_request() -> None: g.lib = app.config["lib"] @@ -298,19 +307,19 @@ def before_request(): @app.route("/item/", methods=["GET", "DELETE", "PATCH"]) @resource("items", patchable=True) -def get_item(id_): +def get_item(id_) -> Any: return g.lib.get_item(id_) @app.route("/item/") @app.route("/item/query/") @resource_list("items") -def all_items(): +def all_items() -> Any: return g.lib.items() @app.route("/item//file") -def item_file(item_id): +def item_file(item_id: int) -> Any: item = g.lib.get_item(item_id) item_path = util.syspath(item.path) @@ -331,12 +340,12 @@ def item_file(item_id): @app.route("/item/query/", methods=["GET", "DELETE", "PATCH"]) @resource_query("items", patchable=True) -def item_query(queries): +def item_query(queries: Sequence[str]) -> Any: return g.lib.items(queries) @app.route("/item/path/") -def item_at_path(path): +def item_at_path(path: str) -> Any: query = PathQuery("path", path.encode("utf-8")) item = g.lib.items(query).get() if item: @@ -345,7 +354,7 @@ def item_at_path(path): @app.route("/item/values/") -def item_unique_field_values(key): +def item_unique_field_values(key: str) -> Any: sort_key = flask.request.args.get("sort_key", key) try: values = _get_unique_table_field_values( @@ -361,25 +370,25 @@ def item_unique_field_values(key): @app.route("/album/", methods=["GET", "DELETE"]) @resource("albums") -def get_album(id_): +def get_album(id_) -> Any: return g.lib.get_album(id_) @app.route("/album/") @app.route("/album/query/") @resource_list("albums") -def all_albums(): +def all_albums() -> Any: return g.lib.albums() @app.route("/album/query/", methods=["GET", "DELETE"]) @resource_query("albums") -def album_query(queries): +def album_query(queries: Sequence[str]) -> Any: return g.lib.albums(queries) @app.route("/album//art") -def album_art(album_id): +def album_art(album_id: int) -> Any: album = g.lib.get_album(album_id) if album and album.artpath: return flask.send_file(album.artpath.decode()) @@ -387,7 +396,7 @@ def album_art(album_id): @app.route("/album/values/") -def album_unique_field_values(key): +def album_unique_field_values(key: str) -> Any: sort_key = flask.request.args.get("sort_key", key) try: values = _get_unique_table_field_values( @@ -402,7 +411,7 @@ def album_unique_field_values(key): @app.route("/artist/") -def all_artists(): +def all_artists() -> Any: with g.lib.transaction() as tx: rows = tx.query("SELECT DISTINCT albumartist FROM albums") all_artists = [row[0] for row in rows] @@ -413,7 +422,7 @@ def all_artists(): @app.route("/stats") -def stats(): +def stats() -> Any: with g.lib.transaction() as tx: item_rows = tx.query("SELECT COUNT(*) FROM items") album_rows = tx.query("SELECT COUNT(*) FROM albums") @@ -424,7 +433,7 @@ def stats(): @app.route("/") -def home(): +def home() -> Any: return flask.render_template("index.html") @@ -432,7 +441,7 @@ def home(): class WebPlugin(BeetsPlugin): - def __init__(self): + def __init__(self) -> None: super().__init__() self.config.add( { @@ -446,7 +455,7 @@ def __init__(self): } ) - def commands(self): + def commands(self) -> list[ui.Subcommand]: cmd = ui.Subcommand("web", help="start a Web interface") cmd.parser.add_option( "-d", @@ -524,10 +533,12 @@ class ReverseProxied: :param app: the WSGI application """ - def __init__(self, app): + def __init__(self, app: t.Callable[..., t.Any]) -> None: self.app = app - def __call__(self, environ, start_response): + def __call__( + self, environ: dict[str, t.Any], start_response: t.Callable[..., t.Any] + ) -> Any: script_name = environ.get("HTTP_X_SCRIPT_NAME", "") if script_name: environ["SCRIPT_NAME"] = script_name From 787f5999c6ec80265210771def26c56b4dd990c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0ar=C5=ABnas=20Nejus?= Date: Wed, 19 Aug 2026 02:19:08 +0100 Subject: [PATCH 2/2] typing: fix API and export plugins --- beetsplug/aura.py | 64 ++++++++++++------ beetsplug/export.py | 136 ++++++++++++++++++++------------------ beetsplug/web/__init__.py | 12 ++-- 3 files changed, 121 insertions(+), 91 deletions(-) diff --git a/beetsplug/aura.py b/beetsplug/aura.py index 2f469731be..2e4dbfe729 100644 --- a/beetsplug/aura.py +++ b/beetsplug/aura.py @@ -27,8 +27,11 @@ from beets.ui import Subcommand, _open_library if TYPE_CHECKING: - from collections.abc import Iterable, Mapping, Sequence + from collections.abc import Iterable, Sequence + from werkzeug.datastructures import MultiDict + + from beets.dbcore import Query from beets.dbcore.query import SQLiteType from beets.dbcore.sort import Sort from beets.library import LibModel, Library @@ -120,9 +123,10 @@ class AURADocument: """Base class for building AURA documents.""" model_cls: ClassVar[type[LibModel]] + attribute_map: ClassVar[dict[str, str]] lib: Library - args: Mapping[str, str] + args: MultiDict[str, str] @classmethod def from_app(cls) -> Self: @@ -166,7 +170,7 @@ def translate_filters(self) -> AndQuery: # filter[]. This regex extracts . pattern = re.compile(r"filter\[(?P[a-zA-Z0-9_-]+)\]") queries = [] - for key, value in self.args.items(): + for key, v in self.args.items(): match = pattern.match(key) if match: # Extract attribute name from key @@ -174,11 +178,11 @@ def translate_filters(self) -> AndQuery: # Get the beets version of the attribute name beets_attr = self.attribute_map.get(aura_attr, aura_attr) converter = self.get_attribute_converter(beets_attr) - value = converter(value) + value = converter(v) # type: ignore[arg-type, misc] # Add exact match query to list # Use a slow query so it works with all fields queries.append( - self.model_cls.field_query(beets_attr, value, MatchQuery) + self.model_cls.field_query(beets_attr, value, MatchQuery) # type: ignore[arg-type] ) # NOTE: AURA doesn't officially support multiple queries return AndQuery(queries) @@ -193,7 +197,7 @@ def translate_sorts(self, sort_arg: str) -> MultipleSort: """ # Change HTTP query parameter to a list aura_sorts = sort_arg.strip(",").split(",") - sorts = [] + sorts: list[Sort] = [] for aura_attr in aura_sorts: if aura_attr[0] == "-": ascending = False @@ -208,6 +212,10 @@ def translate_sorts(self, sort_arg: str) -> MultipleSort: sorts.append(SlowFieldSort(beets_attr, ascending=ascending)) return MultipleSort(sorts) + @staticmethod + def get_resource_object(lib: Library, *args) -> JSONDict | None: + raise NotImplementedError + def paginate( self, collection: Sequence[Any] ) -> tuple[list[Any], str | None]: @@ -252,7 +260,7 @@ def paginate( def get_included( self, data: Iterable[JSONDict], include_str: str - ) -> list[JSONDict | None]: + ) -> list[JSONDict]: """Build a list of resource objects for inclusion. Args: @@ -283,16 +291,16 @@ def get_included( res_type = identifier["type"] if res_type == "track": track_id = int(identifier["id"]) - track = self.lib.get_item(track_id) - included.append( - TrackDocument.get_resource_object(self.lib, track) - ) + if track := self.lib.get_item(track_id): + included.append( + TrackDocument.get_resource_object(self.lib, track) + ) elif res_type == "album": album_id = int(identifier["id"]) - album = self.lib.get_album(album_id) - included.append( - AlbumDocument.get_resource_object(self.lib, album) - ) + if album := self.lib.get_album(album_id): + included.append( + AlbumDocument.get_resource_object(self.lib, album) + ) elif res_type == "artist": artist_id = identifier["id"] included.append( @@ -305,7 +313,14 @@ def get_included( ) else: raise ValueError(f"Invalid resource type: {res_type}") - return included + return list(filter(None, included)) + + def get_collection( + self, + query: str | Sequence[str] | Query | None = None, + sort: Sort | None = None, + ) -> Any: + raise NotImplementedError def all_resources(self) -> JSONDict: """Build document for /tracks, /albums or /artists.""" @@ -317,7 +332,8 @@ def all_resources(self) -> JSONDict: # have a non-empty, non-zero value for that field. query.subqueries.extend( NotQuery( - self.model_cls.field_query(s.field, "(^$|^0$)", RegexpQuery) + # these MultipleSorts have FieldSorts which define .field + self.model_cls.field_query(s.field, "(^$|^0$)", RegexpQuery) # type: ignore[attr-defined] ) for s in sort.sorts ) @@ -362,7 +378,9 @@ class TrackDocument(AURADocument): attribute_map = TRACK_ATTR_MAP def get_collection( - self, query: list[str] | None = None, sort: Sort | None = None + self, + query: str | Sequence[str] | Query | None = None, + sort: Sort | None = None, ) -> Any: """Get Item objects from the library. @@ -444,7 +462,9 @@ class AlbumDocument(AURADocument): attribute_map = ALBUM_ATTR_MAP def get_collection( - self, query: list[str] | None = None, sort: Sort | None = None + self, + query: str | Sequence[str] | Query | None = None, + sort: Sort | None = None, ) -> Any: """Get Album objects from the library. @@ -530,7 +550,9 @@ class ArtistDocument(AURADocument): attribute_map = ARTIST_ATTR_MAP def get_collection( - self, query: list[str] | None = None, sort: Sort | None = None + self, + query: str | Sequence[str] | Query | None = None, + sort: Sort | None = None, ) -> Any: """Get a list of artist names from the library. @@ -921,7 +943,7 @@ def create_app() -> Flask: app.config["lib"] = _open_library(config) # Enable CORS if required - cors = config["aura"]["cors"].as_str_seq(list) + cors = config["aura"]["cors"].as_str_seq(split=True) if cors: from flask_cors import CORS diff --git a/beetsplug/export.py b/beetsplug/export.py index 52660f55a3..f9b906fdab 100644 --- a/beetsplug/export.py +++ b/beetsplug/export.py @@ -5,6 +5,7 @@ import codecs import csv import json +import os import sys from datetime import date, datetime from typing import TYPE_CHECKING, Any, Literal, Protocol, get_args @@ -12,18 +13,21 @@ import mediafile -from beets import ui, util +from beets import ui +from beets.dbcore.types import BasePathType +from beets.library.fields import TYPE_BY_FIELD from beets.plugins import BeetsPlugin from beetsplug.info import library_data, tag_data if TYPE_CHECKING: - from collections.abc import Sequence + from collections.abc import Iterable, Iterator, Sequence from beets.library import Library from ._typing import JSONDict Format = Literal["json", "jsonlines", "csv", "xml"] +VALID_FORMATS = get_args(Format) class ExportCLIOpts(Protocol): @@ -45,6 +49,8 @@ def default(self, o: object) -> Any: class ExportPlugin(BeetsPlugin): + default_format: Format + def __init__(self) -> None: super().__init__() @@ -85,6 +91,9 @@ def __init__(self) -> None: # 'item_fields': [] } ) + self.default_format = self.config["default_format"].as_choice( + VALID_FORMATS + ) def commands(self) -> list[ui.Subcommand]: cmd = ui.Subcommand("export", help="export data from beets") @@ -118,8 +127,10 @@ def commands(self) -> list[ui.Subcommand]: cmd.parser.add_option( "-f", "--format", - default="json", - help="the output format: json (default), jsonlines, csv, or xml", + type="choice", + choices=VALID_FORMATS, + default=self.config["default_format"].get(), + help="the output format: json|jsonlines|csv|xml", ) return [cmd] @@ -128,15 +139,18 @@ def run( ) -> None: file_path = opts.output file_mode = "a" if opts.append else "w" - default_format: Format = self.config["default_format"].as_choice( - get_args(Format) - ) - file_format = opts.format or default_format - file_format_is_line_based = file_format == "jsonlines" + file_format = opts.format or self.default_format format_options = self.config[file_format]["formatting"].get(dict) - export_format = ExportFormat.factory( - file_type=file_format, file_path=file_path, file_mode=file_mode + _format = ( + CSVFormat + if file_format == "csv" + else XMLFormat + if file_format == "xml" + else JsonFormat + ) + export_format = _format( + file_format=file_format, file_path=file_path, file_mode=file_mode ) if opts.library or opts.album: @@ -148,33 +162,43 @@ def run( for keys in opts.included_keys: included_keys.extend(keys.split(",")) - items = [] - for data_emitter in data_collector(lib, args, album=opts.album): - try: - data, _ = data_emitter(included_keys or "*") - except (mediafile.UnreadableFileError, OSError) as ex: - self._log.error("cannot read file: {}", ex) - continue - - for key, value in data.items(): - if isinstance(value, bytes): - data[key] = util.displayable_path(value) - - if file_format_is_line_based: - export_format.export(data, **format_options) - else: - items += [data] - - if not file_format_is_line_based: - export_format.export(items, **format_options) + byte_fields = [ + k for k, v in TYPE_BY_FIELD.items() if isinstance(v, BasePathType) + ] + + def collect_data() -> Iterator[JSONDict]: + for data_emitter in data_collector(lib, args, album=opts.album): + try: + data, _ = data_emitter(included_keys or "*") + except (mediafile.UnreadableFileError, OSError) as ex: + self._log.error("cannot read file: {}", ex) + continue + else: + yield data + + def stringify_bytes(data: JSONDict) -> JSONDict: + for field in byte_fields: + if (value := data.get(field)) is not None: + data[field] = os.fsdecode(value) + + return data + + export_format.export( + map(stringify_bytes, collect_data()), **format_options + ) class ExportFormat: """The output format type""" def __init__( - self, file_path: str, file_mode: str = "w", encoding: str = "utf-8" + self, + file_format: str, + file_path: str | None, + file_mode: str = "w", + encoding: str = "utf-8", ) -> None: + self.file_format = file_format self.path = file_path self.mode = file_mode self.encoding = encoding @@ -185,44 +209,30 @@ def __init__( else sys.stdout ) - @classmethod - def factory( - cls, file_type: str, **kwargs - ) -> JsonFormat | CSVFormat | XMLFormat: - if file_type in ["json", "jsonlines"]: - return JsonFormat(**kwargs) - if file_type == "csv": - return CSVFormat(**kwargs) - if file_type == "xml": - return XMLFormat(**kwargs) - raise NotImplementedError() - - def export(self, data: list[JSONDict], **kwargs) -> None: + def export(self, data_iter: Iterable[JSONDict], **kwargs) -> None: raise NotImplementedError() class JsonFormat(ExportFormat): """Saves in a json file""" - def __init__( - self, file_path: str, file_mode: str = "w", encoding: str = "utf-8" - ) -> None: - super().__init__(file_path, file_mode, encoding) - - def export(self, data: list[JSONDict], **kwargs) -> None: + def _print_json(self, data: Any, **kwargs) -> None: json.dump(data, self.out_stream, cls=ExportEncoder, **kwargs) self.out_stream.write("\n") + def export(self, data_iter: Iterable[JSONDict], **kwargs) -> None: + if self.file_format == "json": + self._print_json(list(data_iter), **kwargs) + else: + for item in data_iter: + self._print_json(item, **kwargs) + class CSVFormat(ExportFormat): """Saves in a csv file""" - def __init__( - self, file_path: str, file_mode: str = "w", encoding: str = "utf-8" - ) -> None: - super().__init__(file_path, file_mode, encoding) - - def export(self, data: list[JSONDict], **kwargs) -> None: + def export(self, data_iter: Iterable[JSONDict], **kwargs) -> None: + data = list(data_iter) header = list(data[0].keys()) if data else [] writer = csv.DictWriter(self.out_stream, fieldnames=header, **kwargs) writer.writeheader() @@ -232,13 +242,9 @@ def export(self, data: list[JSONDict], **kwargs) -> None: class XMLFormat(ExportFormat): """Saves in a xml file""" - def __init__( - self, file_path: str, file_mode: str = "w", encoding: str = "utf-8" - ) -> None: - super().__init__(file_path, file_mode, encoding) - - def export(self, data: list[JSONDict], **kwargs) -> None: + def export(self, data_iter: Iterable[JSONDict], **kwargs) -> None: # Creates the XML file structure. + data = list(data_iter) library = ElementTree.Element("library") tracks = ElementTree.SubElement(library, "tracks") if data and isinstance(data[0], dict): @@ -249,8 +255,8 @@ def export(self, data: list[JSONDict], **kwargs) -> None: track_details.text = value # Depending on the version of python the encoding needs to change try: - data = ElementTree.tostring(library, encoding="unicode", **kwargs) + string = ElementTree.tostring(library, encoding="unicode", **kwargs) except LookupError: - data = ElementTree.tostring(library, encoding="utf-8", **kwargs) + string = ElementTree.tostring(library, encoding="utf-8", **kwargs) - self.out_stream.write(data) + self.out_stream.write(string) diff --git a/beetsplug/web/__init__.py b/beetsplug/web/__init__.py index 3dc75a7371..4543c5472d 100644 --- a/beetsplug/web/__init__.py +++ b/beetsplug/web/__init__.py @@ -126,9 +126,9 @@ def resource(name: str, patchable: bool = False) -> Any: """Decorates a function to handle RESTful HTTP requests for a resource.""" def make_responder(retriever: t.Callable[[int], LibModel | None]) -> Any: - def responder(ids: Sequence[int]): - entities = [retriever(id_) for id_ in ids] - entities = [entity for entity in entities if entity] + def responder(ids: Sequence[int]) -> Any: + retrieved = [retriever(id_) for id_ in ids] + entities = [entity for entity in retrieved if entity] if get_method() == "DELETE": if app.config.get("READONLY", True): @@ -307,7 +307,7 @@ def before_request() -> None: @app.route("/item/", methods=["GET", "DELETE", "PATCH"]) @resource("items", patchable=True) -def get_item(id_) -> Any: +def get_item(id_: int) -> Any: return g.lib.get_item(id_) @@ -321,6 +321,8 @@ def all_items() -> Any: @app.route("/item//file") def item_file(item_id: int) -> Any: item = g.lib.get_item(item_id) + if not item: + return flask.abort(404, f"Item with id {item_id} not found") item_path = util.syspath(item.path) base_filename = os.path.basename(item_path) @@ -370,7 +372,7 @@ def item_unique_field_values(key: str) -> Any: @app.route("/album/", methods=["GET", "DELETE"]) @resource("albums") -def get_album(id_) -> Any: +def get_album(id_: int) -> Any: return g.lib.get_album(id_)