refactor: convert _types from class attributes to cached properties

Convert static _types dictionaries to dynamic cached class properties to
enable proper plugin type inheritance and avoid mutating shared state.

Key changes:
- Replace static _types dicts with @cached_classproperty decorators
- Update cached_classproperty to support proper caching with class names
- Remove manual _types mutation in plugin loading/unloading
- Add pluginload event and cache clearing for proper plugin integration
- Fix test to trigger type checking during item creation

This ensures plugin types are properly inherited through the class
hierarchy and eliminates issues with shared mutable state between
test runs.
This commit is contained in:
Šarūnas Nejus 2025-07-15 11:58:25 +01:00
parent 852cbd2650
commit 3be4a89aee
No known key found for this signature in database
GPG key ID: DD28F6704DBE3435
8 changed files with 60 additions and 38 deletions

View file

@ -289,9 +289,10 @@ class Model(ABC, Generic[D]):
terms. terms.
""" """
_types: dict[str, types.Type] = {} @cached_classproperty
"""Optional Types for non-fixed (i.e., flexible and computed) fields. def _types(cls) -> dict[str, types.Type]:
""" """Optional types for non-fixed (flexible and computed) fields."""
return {}
_sorts: dict[str, type[FieldSort]] = {} _sorts: dict[str, type[FieldSort]] = {}
"""Optional named sort criteria. The keys are strings and the values """Optional named sort criteria. The keys are strings and the values

View file

@ -41,6 +41,14 @@ class LibModel(dbcore.Model["Library"]):
_format_config_key: str _format_config_key: str
path: bytes path: bytes
@cached_classproperty
def _types(cls) -> dict[str, types.Type]:
"""Return the types of the fields in this model."""
return {
**plugins.types(cls), # type: ignore[arg-type]
"data_source": types.STRING,
}
@cached_classproperty @cached_classproperty
def writable_media_fields(cls) -> set[str]: def writable_media_fields(cls) -> set[str]:
return set(MediaFile.fields()) & cls._fields.keys() return set(MediaFile.fields()) & cls._fields.keys()
@ -265,10 +273,9 @@ class Album(LibModel):
_search_fields = ("album", "albumartist", "genre") _search_fields = ("album", "albumartist", "genre")
_types = { @cached_classproperty
"path": types.PathType(), def _types(cls) -> dict[str, types.Type]:
"data_source": types.STRING, return {**super()._types, "path": types.PathType()}
}
_sorts = { _sorts = {
"albumartist": dbcore.query.SmartArtistSort, "albumartist": dbcore.query.SmartArtistSort,
@ -715,10 +722,6 @@ class Item(LibModel):
"genre", "genre",
) )
_types = {
"data_source": types.STRING,
}
# Set of item fields that are backed by `MediaFile` fields. # Set of item fields that are backed by `MediaFile` fields.
# Any kind of field (fixed, flexible, and computed) may be a media # Any kind of field (fixed, flexible, and computed) may be a media
# field. Only these fields are read from disk in `read` and written in # field. Only these fields are read from disk in `read` and written in

View file

@ -362,7 +362,7 @@ def queries() -> dict[str, type[Query]]:
def types(model_cls: type[AnyModel]) -> dict[str, Type]: def types(model_cls: type[AnyModel]) -> dict[str, Type]:
# Gives us `item_types` and `album_types` """Return mapping between flex field names and types for the given model."""
attr_name = f"{model_cls.__name__.lower()}_types" attr_name = f"{model_cls.__name__.lower()}_types"
types: dict[str, Type] = {} types: dict[str, Type] = {}
for plugin in find_plugins(): for plugin in find_plugins():

View file

@ -58,6 +58,7 @@ from beets.ui.commands import TerminalImportSession
from beets.util import ( from beets.util import (
MoveOperation, MoveOperation,
bytestring_path, bytestring_path,
cached_classproperty,
clean_module_tempdir, clean_module_tempdir,
syspath, syspath,
) )
@ -471,8 +472,6 @@ class PluginMixin(ConfigMixin):
plugin: ClassVar[str] plugin: ClassVar[str]
preload_plugin: ClassVar[bool] = True preload_plugin: ClassVar[bool] = True
original_item_types = dict(Item._types)
original_album_types = dict(Album._types)
original_item_queries = dict(Item._queries) original_item_queries = dict(Item._queries)
original_album_queries = dict(Album._queries) original_album_queries = dict(Album._queries)
@ -494,13 +493,12 @@ class PluginMixin(ConfigMixin):
# FIXME this should eventually be handled by a plugin manager # FIXME this should eventually be handled by a plugin manager
plugins = (self.plugin,) if hasattr(self, "plugin") else plugins plugins = (self.plugin,) if hasattr(self, "plugin") else plugins
self.config["plugins"] = plugins self.config["plugins"] = plugins
cached_classproperty.cache.clear()
beets.plugins.load_plugins(plugins) beets.plugins.load_plugins(plugins)
beets.plugins.send("pluginload")
beets.plugins.find_plugins() beets.plugins.find_plugins()
# Take a backup of the original _types and _queries to restore # Take a backup of the original _queries to restore when unloading.
# when unloading.
Item._types.update(beets.plugins.types(Item))
Album._types.update(beets.plugins.types(Album))
Item._queries.update(beets.plugins.named_queries(Item)) Item._queries.update(beets.plugins.named_queries(Item))
Album._queries.update(beets.plugins.named_queries(Album)) Album._queries.update(beets.plugins.named_queries(Album))
@ -512,8 +510,6 @@ class PluginMixin(ConfigMixin):
self.config["plugins"] = [] self.config["plugins"] = []
beets.plugins._classes = set() beets.plugins._classes = set()
beets.plugins._instances = {} beets.plugins._instances = {}
Item._types = self.original_item_types
Album._types = self.original_album_types
Item._queries = self.original_item_queries Item._queries = self.original_item_queries
Album._queries = self.original_album_queries Album._queries = self.original_album_queries

View file

@ -1609,14 +1609,7 @@ def _setup(options, lib=None):
plugins = _load_plugins(options, config) plugins = _load_plugins(options, config)
# Add types and queries defined by plugins. # Add queries defined by plugins.
plugin_types_album = plugins.types(library.Album)
library.Album._types.update(plugin_types_album)
item_types = plugin_types_album.copy()
item_types.update(library.Item._types)
item_types.update(plugins.types(library.Item))
library.Item._types = item_types
library.Item._queries.update(plugins.named_queries(library.Item)) library.Item._queries.update(plugins.named_queries(library.Item))
library.Album._queries.update(plugins.named_queries(library.Album)) library.Album._queries.update(plugins.named_queries(library.Album))

View file

@ -41,6 +41,7 @@ from typing import (
Any, Any,
AnyStr, AnyStr,
Callable, Callable,
ClassVar,
Generic, Generic,
NamedTuple, NamedTuple,
TypeVar, TypeVar,
@ -1052,20 +1053,46 @@ def par_map(transform: Callable[[T], Any], items: Sequence[T]) -> None:
class cached_classproperty: class cached_classproperty:
"""A decorator implementing a read-only property that is *lazy* in """Descriptor implementing cached class properties.
the sense that the getter is only invoked once. Subsequent accesses
through *any* instance use the cached result. Provides class-level dynamic property behavior where the getter function is
called once per class and the result is cached for subsequent access. Unlike
instance properties, this operates on the class rather than instances.
""" """
def __init__(self, getter): cache: ClassVar[dict[tuple[Any, str], Any]] = {}
name: str
# Ideally, we would like to use `Callable[[type[T]], Any]` here,
# however, `mypy` is unable to see this as a **class** property, and thinks
# that this callable receives an **instance** of the object, failing the
# type check, for example:
# >>> class Album:
# >>> @cached_classproperty
# >>> def foo(cls):
# >>> reveal_type(cls) # mypy: revealed type is "Album"
# >>> return cls.bar
#
# Argument 1 to "cached_classproperty" has incompatible type
# "Callable[[Album], ...]"; expected "Callable[[type[Album]], ...]"
#
# Therefore, we just use `Any` here, which is not ideal, but works.
def __init__(self, getter: Callable[[Any], Any]) -> None:
"""Initialize the descriptor with the property getter function."""
self.getter = getter self.getter = getter
self.cache = {}
def __get__(self, instance, owner): def __set_name__(self, owner: Any, name: str) -> None:
if owner not in self.cache: """Capture the attribute name this descriptor is assigned to."""
self.cache[owner] = self.getter(owner) self.name = name
return self.cache[owner] def __get__(self, instance: Any, owner: type[Any]) -> Any:
"""Compute and cache if needed, and return the property value."""
key = owner, self.name
if key not in self.cache:
self.cache[key] = self.getter(owner)
return self.cache[key]
class LazySharedInstance(Generic[T]): class LazySharedInstance(Generic[T]):

View file

@ -58,7 +58,9 @@ class AdvancedRewritePlugin(BeetsPlugin):
def __init__(self): def __init__(self):
"""Parse configuration and register template fields for rewriting.""" """Parse configuration and register template fields for rewriting."""
super().__init__() super().__init__()
self.register_listener("pluginload", self.loaded)
def loaded(self):
template = confuse.Sequence( template = confuse.Sequence(
confuse.OneOf( confuse.OneOf(
[ [

View file

@ -134,7 +134,7 @@ class TypesPluginTest(PluginTestCase):
def test_unknown_type_error(self): def test_unknown_type_error(self):
self.config["types"] = {"flex": "unkown type"} self.config["types"] = {"flex": "unkown type"}
with pytest.raises(ConfigValueError): with pytest.raises(ConfigValueError):
self.run_command("ls") self.add_item(flex="test")
def test_template_if_def(self): def test_template_if_def(self):
# Tests for a subtle bug when using %ifdef in templates along with # Tests for a subtle bug when using %ifdef in templates along with