From 8d03803fcf1b18385e477cb9273553e2431447ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0ar=C5=ABnas=20Nejus?= Date: Tue, 7 Jul 2026 20:39:54 +0100 Subject: [PATCH] typing: type beets.dbcore.db --- beets/dbcore/db.py | 86 ++++++++++++++++++++------------------ beets/library/models.py | 7 ++-- beets/util/functemplate.py | 4 +- 3 files changed, 53 insertions(+), 44 deletions(-) diff --git a/beets/dbcore/db.py b/beets/dbcore/db.py index 78c443239..873c81d31 100755 --- a/beets/dbcore/db.py +++ b/beets/dbcore/db.py @@ -52,6 +52,8 @@ if TYPE_CHECKING: from sqlite3 import Connection from types import TracebackType + from beets.util import PathLike + from .query import FieldQueryType, Query, SQLiteType from .sort import FieldSort, Sort @@ -74,7 +76,7 @@ class DBAccessError(Exception): class DBCustomFunctionError(Exception): """A sqlite function registered by beets failed.""" - def __init__(self): + def __init__(self) -> None: super().__init__( "beets defined SQLite function failed; " "see the other errors above for details" @@ -183,7 +185,7 @@ class LazyDict(UserDict[str, Any]): if key in self._raw: del self._raw[key] - def __contains__(self, key: Any) -> bool: + def __contains__(self, key: object) -> bool: """Determine whether `key` is an attribute on this object.""" return key in self.data or key in self._raw @@ -294,7 +296,7 @@ class Model(ABC, Generic[D]): """ @cached_classproperty - def _relation(cls): + def _relation(cls) -> type[Model]: """The model that this model is closely related to.""" return cls @@ -336,7 +338,7 @@ class Model(ABC, Generic[D]): raise NotFoundError(f"No matching {model_cls.__name__} found") from None @classmethod - def _getters(cls: type[Model]): + def _getters(cls) -> dict[str, Callable[[Self], object]]: """Return a mapping from field names to getter functions.""" # We could cache this if it becomes a performance problem to # gather the getter mapping every time. @@ -356,7 +358,7 @@ class Model(ABC, Generic[D]): db: D | None = None, fixed_values: JSONDict | None = None, flex_values: JSONDict | None = None, - **kwargs, + **kwargs: Any, ) -> None: """Create a new model instance. @@ -378,7 +380,7 @@ class Model(ABC, Generic[D]): f"({', '.join(f'{k}={v!r}' for k, v in dict(self).items())})" ) - def clear_dirty(self): + def clear_dirty(self) -> None: """Mark all fields as *clean* (i.e., not needing to be stored to the database). Also update the revision. """ @@ -416,7 +418,7 @@ class Model(ABC, Generic[D]): # Essential field accessors. @classmethod - def _type(cls, key) -> types.Type: + def _type(cls, key: str) -> types.Type: """Get the type of a field, a `Type` instance. If the field has no explicit type, it is given the base `Type`, @@ -444,7 +446,7 @@ class Model(ABC, Generic[D]): """Convert the attribute type according to the SQL type""" return cls._type(key).from_sql(value) - def _get(self, key, default: Any = None, raise_: bool = False): + def _get(self, key: str, default: Any = None, raise_: bool = False) -> Any: """Get the value for a field, or `default`. Alternatively, raise a KeyError if the field is not available. """ @@ -463,13 +465,13 @@ class Model(ABC, Generic[D]): get = _get - def __getitem__(self, key): + def __getitem__(self, key: str) -> Any: """Get the value for a field. Raise a KeyError if the field is not available. """ return self._get(key, raise_=True) - def _setitem(self, key, value): + def _setitem(self, key: str, value: Any) -> bool: """Assign the value for a field, return whether new and old value differ. """ @@ -491,11 +493,11 @@ class Model(ABC, Generic[D]): return changed - def __setitem__(self, key, value): + def __setitem__(self, key: str, value: Any) -> None: """Assign the value for a field.""" self._setitem(key, value) - def __delitem__(self, key): + def __delitem__(self, key: str) -> None: """Remove a flexible attribute from the model.""" if key in self._values_flex: # Flexible. del self._values_flex[key] @@ -527,7 +529,7 @@ class Model(ABC, Generic[D]): # Act like a dictionary. - def update(self, values): + def update(self, values: Mapping[str, Any]) -> None: """Assign all values in the given dict.""" for key, value in values.items(): self[key] = value @@ -539,7 +541,7 @@ class Model(ABC, Generic[D]): for key in self: yield key, self[key] - def __contains__(self, key) -> bool: + def __contains__(self, key: str) -> bool: """Determine whether `key` is an attribute on this object.""" return key in self.keys(computed=True) @@ -551,7 +553,7 @@ class Model(ABC, Generic[D]): # Convenient attribute access. - def __getattr__(self, key): + def __getattr__(self, key: str) -> Any: if key.startswith("_"): raise AttributeError(f"model has no attribute {key!r}") try: @@ -559,13 +561,13 @@ class Model(ABC, Generic[D]): except KeyError: raise AttributeError(f"no such field {key!r}") - def __setattr__(self, key, value): + def __setattr__(self, key: str, value: Any) -> None: if key.startswith("_"): super().__setattr__(key, value) else: self[key] = value - def __delattr__(self, key): + def __delattr__(self, key: str) -> None: if key.startswith("_"): super().__delattr__(key) else: @@ -573,7 +575,7 @@ class Model(ABC, Generic[D]): # Database interaction (CRUD methods). - def store(self, fields: Iterable[str] | None = None): + def store(self, fields: Iterable[str] | None = None) -> None: """Save the object's metadata into the library database. :param fields: the fields to be stored. If not specified, all fields will be. @@ -619,7 +621,7 @@ class Model(ABC, Generic[D]): self.clear_dirty() - def load(self): + def load(self) -> None: """Refresh the object's metadata from the library database. If check_revision is true, the database is only queried loaded when a @@ -632,7 +634,7 @@ class Model(ABC, Generic[D]): self.__dict__.update(self.get_fresh_from_db().__dict__) self.clear_dirty() - def remove(self): + def remove(self) -> None: """Remove the object's associated rows from the database.""" with self.db.transaction() as tx: tx.mutate(f"DELETE FROM {self._table} WHERE id=?", (self.id,)) @@ -696,18 +698,18 @@ class Model(ABC, Generic[D]): # Parsing. @classmethod - def _parse(cls, key, string: str) -> Any: + def _parse(cls, key: str, string: str) -> Any: """Parse a string as a value for the given key.""" if not isinstance(string, str): raise TypeError("_parse() argument must be a string") return cls._type(key).parse(string) - def set_parse(self, key, string: str): + def set_parse(self, key: str, string: str) -> None: """Set the object's key to a value represented by a string.""" self[key] = self._parse(key, string) - def __getstate__(self): + def __getstate__(self) -> JSONDict: """Return the state of the object for pickling. Remove the database connection as sqlite connections are not picklable. @@ -733,10 +735,10 @@ class Results(Generic[AnyModel]): model_class: type[AnyModel], rows: list[sqlite3.Row], db: D, - flex_rows, + flex_rows: list[sqlite3.Row], query: Query | None = None, - sort=None, - ): + sort: Sort | None = None, + ) -> None: """Create a result set that will construct objects of type `model_class`. @@ -862,7 +864,7 @@ class Results(Generic[AnyModel]): """Does this result contain any objects?""" return bool(len(self)) - def __getitem__(self, n): + def __getitem__(self, n: int) -> AnyModel: """Get the nth item in this result set. This is inefficient: all items up to n are materialized and thrown away. """ @@ -900,7 +902,7 @@ class Transaction: current transaction. """ - def __init__(self, db: Database): + def __init__(self, db: Database) -> None: self.db = db def __enter__(self) -> Transaction: @@ -982,21 +984,23 @@ class Transaction: else: self._mutated = True - def mutate(self, statement: str, subvals: Sequence[SQLiteType] = ()) -> Any: + def mutate( + self, statement: str, subvals: Sequence[SQLiteType] = () + ) -> int | None: """Run one write statement with shared mutation/error handling.""" with self._handle_mutate(): return self.db._connection().execute(statement, subvals).lastrowid def mutate_many( self, statement: str, subvals: Sequence[tuple[SQLiteType, ...]] = () - ) -> Any: + ) -> int | None: """Run batched writes with shared mutation/error handling.""" with self._handle_mutate(): return ( self.db._connection().executemany(statement, subvals).lastrowid ) - def script(self, statements: str): + def script(self, statements: str) -> None: """Execute a string containing multiple SQL statements.""" # We don't know whether this mutates, but quite likely it does. self._mutated = True @@ -1027,7 +1031,9 @@ class Migration(ABC): finally: self.db._connection().row_factory = original_factory - def migrate_model(self, model_cls: type[Model], *args, **kwargs) -> None: + def migrate_model( + self, model_cls: type[Model], *args: Any, **kwargs: Any + ) -> None: """Run this migration once for a model's backing table.""" table = model_cls._table if not self.db.migration_exists(self.name, table): @@ -1076,7 +1082,7 @@ class Database: data is written in a transaction. """ - def __init__(self, path, timeout: float = 5.0): + def __init__(self, path: PathLike, timeout: float = 5.0) -> None: if sqlite3.threadsafety == 0: raise RuntimeError( "sqlite3 must be compiled with multi-threading support" @@ -1204,8 +1210,8 @@ class Database: conn.row_factory = sqlite3.Row return conn - def add_functions(self, conn): - def regexp(value, pattern): + def add_functions(self, conn: sqlite3.Connection) -> None: + def regexp(value: Any, pattern: str) -> bool: if isinstance(value, bytes): value = value.decode() return re.search(pattern, str(value)) is not None @@ -1234,7 +1240,7 @@ class Database: create_function("unidecode", 1, unidecode) create_function("bytelower", 1, bytelower) - def _close(self): + def _close(self) -> None: """Close the all connections to the underlying SQLite database from all threads. This does not render the database object unusable; new connections can still be opened on demand. @@ -1265,7 +1271,7 @@ class Database: """ return Transaction(self) - def load_extension(self, path: str): + def load_extension(self, path: str) -> None: """Load an SQLite extension into all open connections.""" if not self.supports_extensions: raise ValueError( @@ -1280,7 +1286,7 @@ class Database: # Schema setup and migration. - def _make_table(self, table: str, fields: Mapping[str, types.Type]): + def _make_table(self, table: str, fields: Mapping[str, types.Type]) -> None: """Set up the schema of the database. `fields` is a mapping from field names to `Type`s. Columns are added if necessary. """ @@ -1304,7 +1310,7 @@ class Database: with self.transaction() as tx: tx.script(setup_sql) - def _make_attribute_table(self, flex_table: str): + def _make_attribute_table(self, flex_table: str) -> None: """Create a table and associated index for flexible attributes for the given entity (if they don't exist). """ @@ -1320,7 +1326,7 @@ class Database: ON {flex_table} (entity_id); """) - def _create_indices(self, table: str, indices: Sequence[Index]): + def _create_indices(self, table: str, indices: Sequence[Index]) -> None: """Create indices for the given table if they don't exist.""" with self.transaction() as tx: for index in indices: diff --git a/beets/library/models.py b/beets/library/models.py index d126de83b..95cd30c13 100644 --- a/beets/library/models.py +++ b/beets/library/models.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import TYPE_CHECKING, ClassVar, TypeVar from mediafile import MediaFile, UnreadableFileError +from typing_extensions import Self import beets from beets import dbcore, logging, plugins, util @@ -33,7 +34,7 @@ from .fields import TYPE_BY_FIELD from .queries import parse_query_string if TYPE_CHECKING: - from collections.abc import KeysView + from collections.abc import Callable, KeysView from beets.dbcore.query import FieldQuery, FieldQueryType from beets.dbcore.sort import FieldSort @@ -348,7 +349,7 @@ class Album(LibModel): return Path(os.fsdecode(self.artpath)) if self.artpath else None @classmethod - def _getters(cls): + def _getters(cls) -> dict[str, Callable[[Self], object]]: # In addition to plugin-provided computed fields, also expose # the album's directory as `path`. getters = plugins.album_field_getters() @@ -764,7 +765,7 @@ class Item(LibModel): self.__album = album @classmethod - def _getters(cls): + def _getters(cls) -> dict[str, Callable[[Self], object]]: getters = plugins.item_field_getters() getters["singleton"] = lambda i: i.album_id is None getters["filesize"] = Item.try_filesize # In bytes. diff --git a/beets/util/functemplate.py b/beets/util/functemplate.py index a3541ab34..5547bcb0e 100644 --- a/beets/util/functemplate.py +++ b/beets/util/functemplate.py @@ -12,6 +12,8 @@ This is sort of like a tiny, horrible degeneration of a real templating engine like Jinja2 or Mustache. """ +from __future__ import annotations + import ast import dis import functools @@ -495,7 +497,7 @@ def _parse(template): @functools.lru_cache(maxsize=128) -def template(fmt): +def template(fmt) -> Template: return Template(fmt)