music-assistant-server

1"""Helpers for dealing with API's to interact with Music Assistant."""
2
3from __future__ import annotations
4
5import importlib
6import inspect
7import logging
8import pkgutil
9from collections.abc import AsyncGenerator, Callable, Coroutine, Iterable, Sequence
10from dataclasses import MISSING, dataclass
11from datetime import datetime
12from enum import Enum
13from functools import cache
14from types import NoneType, UnionType
15from typing import TYPE_CHECKING, Any, TypeVar, Union, get_args, get_origin, get_type_hints
16
17from mashumaro.exceptions import MissingField
18from music_assistant_models.media_items.media_item import MediaItem
19
20from music_assistant.helpers.util import try_parse_bool
21
22if TYPE_CHECKING:
23    from music_assistant_models.auth import Scope
24
25LOGGER = logging.getLogger(__name__)
26
27_F = TypeVar("_F", bound=Callable[..., Any])
28
29# Cache for resolved type alias strings to avoid repeated imports
30_TYPE_ALIAS_CACHE: dict[str, Any] = {}
31_MODEL_TYPE_CACHE: dict[str, Any] = {}
32_MAX_TYPE_HINT_RESOLVE_ATTEMPTS = 32
33
34
35def _resolve_string_type(type_str: str) -> Any:
36    """
37    Resolve a string type reference back to the actual type.
38
39    This is needed when type aliases like ConfigValueType are converted to strings
40    during type hint resolution to avoid isinstance() errors with complex unions.
41
42    Uses a module-level cache to avoid repeated imports.
43
44    :param type_str: String name of the type (e.g., "ConfigValueType").
45    :return: The actual type object, or the string if resolution fails.
46    """
47    # Check cache first
48    if type_str in _TYPE_ALIAS_CACHE:
49        return _TYPE_ALIAS_CACHE[type_str]
50
51    type_alias_map = {
52        "ConfigValueType": ("music_assistant_models.config_entries", "ConfigValueType"),
53        "MediaItemType": ("music_assistant_models.media_items", "MediaItemType"),
54    }
55
56    if type_str not in type_alias_map:
57        # Cache the string itself for unknown types
58        _TYPE_ALIAS_CACHE[type_str] = type_str
59        return type_str
60
61    module_name, type_name = type_alias_map[type_str]
62    try:
63        module = importlib.import_module(module_name)
64        resolved_type = getattr(module, type_name)
65        # Cache the successfully resolved type
66        _TYPE_ALIAS_CACHE[type_str] = resolved_type
67        return resolved_type
68    except (ImportError, AttributeError) as err:
69        LOGGER.warning("Failed to resolve type alias %s: %s", type_str, err)
70        # Cache the string to avoid repeated failed attempts
71        _TYPE_ALIAS_CACHE[type_str] = type_str
72        return type_str
73
74
75@cache
76def _get_model_module_names() -> tuple[str, ...]:
77    """Return all module names from the music_assistant_models package."""
78    try:
79        music_assistant_models = importlib.import_module("music_assistant_models")
80    except ImportError:
81        return ()
82    try:
83        return tuple(
84            mod.name
85            for mod in pkgutil.walk_packages(
86                music_assistant_models.__path__, prefix="music_assistant_models."
87            )
88        )
89    except Exception:
90        return ()
91
92
93def _resolve_model_type(name: str) -> Any | None:
94    """Resolve a type from the music_assistant_models package by name."""
95    if name in _MODEL_TYPE_CACHE:
96        return _MODEL_TYPE_CACHE[name]
97
98    try:
99        music_assistant_models = importlib.import_module("music_assistant_models")
100    except ImportError:
101        _MODEL_TYPE_CACHE[name] = None
102        return None
103
104    if hasattr(music_assistant_models, name):
105        resolved = getattr(music_assistant_models, name)
106        _MODEL_TYPE_CACHE[name] = resolved
107        return resolved
108
109    for module_name in _get_model_module_names():
110        try:
111            module = importlib.import_module(module_name)
112            if hasattr(module, name):
113                resolved = getattr(module, name)
114                _MODEL_TYPE_CACHE[name] = resolved
115                return resolved
116        except ImportError:
117            continue
118
119    _MODEL_TYPE_CACHE[name] = None
120    return None
121
122
123def _extract_name_error_symbol(error: NameError) -> str | None:
124    """Extract the missing symbol name from NameError messages."""
125    msg = str(error)
126    if "name '" in msg and "' is not defined" in msg:
127        return msg.split("name '", 1)[1].split("'", 1)[0]
128    return None
129
130
131def _get_type_hints_for_api_command(func: Callable[..., Any]) -> dict[str, Any]:
132    """
133    Get type hints for API command handlers with fallback for model types.
134
135    We need this because API command handlers are often declared in controllers with
136    type-only imports under TYPE_CHECKING (to avoid runtime cycles), e.g.:
137      from typing import TYPE_CHECKING
138      if TYPE_CHECKING:
139          from music_assistant_models.background_task import BackgroundTask
140
141    Without this fallback, get_type_hints() raises NameError when evaluating
142    forward refs like "BackgroundTask" at runtime during API registration.
143    """
144    globalns = dict(getattr(func, "__globals__", {}))
145    localns: dict[str, Any] = {}
146    for _attempt in range(_MAX_TYPE_HINT_RESOLVE_ATTEMPTS):
147        try:
148            return get_type_hints(func, globalns=globalns, localns=localns)
149        except NameError as err:
150            missing_name = _extract_name_error_symbol(err)
151            if not missing_name:
152                raise
153            resolved = _resolve_model_type(missing_name)
154            if resolved is None:
155                raise
156            globalns[missing_name] = resolved
157            continue
158    msg = f"Exceeded type hint resolution attempts for API command {func.__qualname__}"
159    raise RuntimeError(msg)
160
161
162def _resolve_generic_type_args(
163    args: tuple[Any, ...],
164    func: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]],
165    config_value_type: Any,
166    media_item_type: Any,
167) -> tuple[list[Any], bool]:
168    """
169    Resolve TypeVars and type aliases in generic type arguments.
170
171    :param args: Type arguments from a generic type (e.g., from list[T] or dict[K, V])
172    :param func: The function being analyzed
173    :param config_value_type: The ConfigValueType type alias to compare against
174    :param media_item_type: The MediaItemType type alias to compare against
175    :return: Tuple of (resolved_args, changed) where changed is True if any args were modified
176    """
177    new_args: list[Any] = []
178    changed = False
179
180    for arg in args:
181        # Check if arg matches ConfigValueType union (type alias that was expanded)
182        if arg == config_value_type:
183            # Replace with string reference to preserve type alias
184            new_args.append("ConfigValueType")
185            changed = True
186        # Check if arg matches MediaItemType union (type alias that was expanded)
187        elif arg == media_item_type:
188            # Replace with string reference to preserve type alias
189            new_args.append("MediaItemType")
190            changed = True
191        elif isinstance(arg, TypeVar):
192            # For ItemCls, resolve to concrete type
193            if arg.__name__ == "ItemCls" and hasattr(func, "__self__"):
194                if hasattr(func.__self__, "item_cls"):
195                    new_args.append(func.__self__.item_cls)
196                    changed = True
197                else:
198                    new_args.append(arg)
199            # For ConfigValue TypeVars, resolve to string name
200            elif "ConfigValue" in arg.__name__:
201                new_args.append("ConfigValueType")
202                changed = True
203            else:
204                new_args.append(arg)
205        # Check if arg is a Union containing a TypeVar
206        elif get_origin(arg) in (Union, UnionType):
207            union_args = get_args(arg)
208            for union_arg in union_args:
209                if isinstance(union_arg, TypeVar) and union_arg.__bound__ is not None:
210                    # Resolve the TypeVar in the union
211                    union_arg_index = union_args.index(union_arg)
212                    resolved = _resolve_typevar_in_union(
213                        union_arg, func, union_args, union_arg_index
214                    )
215                    new_args.append(resolved)
216                    changed = True
217                    break
218            else:
219                # No TypeVar found in union, keep as-is
220                new_args.append(arg)
221        else:
222            new_args.append(arg)
223
224    return new_args, changed
225
226
227def _resolve_typevar_in_union(
228    arg: TypeVar,
229    func: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]],
230    args: tuple[Any, ...],
231    i: int,
232) -> Any:
233    """
234    Resolve a TypeVar found in a Union to its concrete type.
235
236    :param arg: The TypeVar to resolve.
237    :param func: The function being analyzed.
238    :param args: All args from the Union.
239    :param i: Index of the TypeVar in the args.
240    """
241    bound_type = arg.__bound__
242    if not bound_type or not hasattr(arg, "__name__"):
243        return bound_type
244
245    type_var_name = arg.__name__
246
247    # Map TypeVar names to their type alias names
248    if "ConfigValue" in type_var_name:
249        return "ConfigValueType"
250
251    if type_var_name == "ItemCls":
252        # Resolve ItemCls to the actual media item class (e.g., Artist, Album, Track)
253        if hasattr(func, "__self__") and hasattr(func.__self__, "item_cls"):
254            resolved_type = func.__self__.item_cls
255            # Preserve other types in the union (like None for Optional)
256            other_args = [a for j, a in enumerate(args) if j != i]
257            if other_args:
258                # Reconstruct union with resolved type
259                return Union[resolved_type, *other_args]
260            return resolved_type
261        # Fallback to bound if we can't get item_cls
262        return bound_type
263
264    # Check if the bound is MediaItemType by comparing the union
265    from music_assistant_models.media_items import (  # noqa: PLC0415
266        MediaItemType as media_item_type,  # noqa: N813
267    )
268
269    if bound_type == media_item_type:
270        return "MediaItemType"
271
272    # Fallback to the bound type
273    return bound_type
274
275
276@dataclass
277class APICommandHandler:
278    """Model for an API command handler."""
279
280    command: str
281    signature: inspect.Signature
282    type_hints: dict[str, Any]
283    target: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]]
284    authenticated: bool = True
285    required_scope: Scope | None = None  # None means any authenticated user
286    allow_impersonation: bool = False  # If True, the command accepts a 'user' argument
287    alias: bool = False  # If True, this is an alias for backward compatibility
288
289    @classmethod
290    def parse(
291        cls,
292        command: str,
293        func: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]],
294        authenticated: bool = True,
295        required_scope: Scope | None = None,
296        allow_impersonation: bool = False,
297        alias: bool = False,
298    ) -> APICommandHandler:
299        """
300        Parse APICommandHandler by providing a function.
301
302        :param command: The command name/path.
303        :param func: The function to handle the command.
304        :param authenticated: Whether authentication is required (default: True).
305        :param required_scope: Scope required to execute the command,
306            None for any authenticated user.
307        :param allow_impersonation: Whether the command accepts a 'user' argument
308            to execute the command on behalf of another user (default: False).
309        :param alias: Whether this is an alias for backward compatibility (default: False).
310        """
311        type_hints = _get_type_hints_for_api_command(func)
312        # workaround for generic typevar ItemCls that needs to be resolved
313        # to the real media item type. TODO: find a better way to do this
314        # without this hack
315        # Import type aliases to compare against
316        from music_assistant_models.config_entries import (  # noqa: PLC0415
317            ConfigValueType as config_value_type,  # noqa: N813
318        )
319        from music_assistant_models.media_items import (  # noqa: PLC0415
320            MediaItemType as media_item_type,  # noqa: N813
321        )
322
323        for key, value in type_hints.items():
324            # Handle generic types (list, tuple, dict, etc.) that may contain TypeVars
325            # For example: list[ItemCls] should become list[Artist]
326            # For example: dict[str, ConfigValueType] should preserve ConfigValueType
327            origin = get_origin(value)
328            if origin in (list, tuple, set, frozenset, dict):
329                args = get_args(value)
330                if args:
331                    new_args, changed = _resolve_generic_type_args(
332                        args, func, config_value_type, media_item_type
333                    )
334                    if changed:
335                        # Reconstruct the generic type with resolved TypeVars
336                        type_hints[key] = origin[tuple(new_args)]
337                continue
338
339            # Handle Union types that may contain TypeVars
340            # For example: _ConfigValueT | ConfigValueType should become just "ConfigValueType"
341            # when _ConfigValueT is bound to ConfigValueType
342            if origin is Union or origin is UnionType:
343                args = get_args(value)
344                # Check if union contains a TypeVar
345                # If the TypeVar's bound is a union that was flattened into the current union,
346                # we can just use the bound type for documentation purposes
347                typevar_found = False
348                for i, arg in enumerate(args):
349                    if isinstance(arg, TypeVar) and arg.__bound__ is not None:
350                        typevar_found = True
351                        type_hints[key] = _resolve_typevar_in_union(arg, func, args, i)
352                        break
353                if typevar_found:
354                    continue
355            if not hasattr(value, "__name__"):
356                continue
357            if value.__name__ == "ItemCls":
358                type_hints[key] = func.__self__.item_cls  # type: ignore[attr-defined]
359            # Resolve TypeVars to their bound type for API documentation
360            # This handles cases like _ConfigValueT which should show as ConfigValueType
361            elif isinstance(value, TypeVar):
362                if value.__bound__ is not None:
363                    type_hints[key] = value.__bound__
364        signature = inspect.signature(func)
365        # the 'user' argument of impersonation-enabled commands is injected by the
366        # command dispatch, so it may not clash with the handler's own arguments
367        if allow_impersonation and not signature.parameters.keys().isdisjoint(("user", "username")):
368            msg = f"Command {command} allows impersonation but accepts a user(name) argument"
369            raise RuntimeError(msg)
370        return APICommandHandler(
371            command=command,
372            signature=signature,
373            type_hints=type_hints,
374            target=func,
375            authenticated=authenticated,
376            required_scope=required_scope,
377            allow_impersonation=allow_impersonation,
378            alias=alias,
379        )
380
381
382def api_command(
383    command: str,
384    authenticated: bool = True,
385    required_scope: Scope | None = None,
386    allow_impersonation: bool = False,
387    alias: bool = False,
388) -> Callable[[_F], _F]:
389    """
390    Decorate a function as API route/command.
391
392    :param command: The command name/path.
393    :param authenticated: Whether authentication is required (default: True).
394    :param required_scope: Scope required to execute the command,
395        None means any authenticated user.
396    :param allow_impersonation: Whether the command accepts a 'user' argument
397        to execute the command on behalf of another user (default: False).
398    :param alias: Whether this is a backward-compatible alias (default: False).
399        Aliases remain functional but are hidden from the API documentation.
400    """
401
402    def decorate(func: _F) -> _F:
403        func.api_cmd = command  # type: ignore[attr-defined]
404        func.api_authenticated = authenticated  # type: ignore[attr-defined]
405        func.api_required_scope = required_scope  # type: ignore[attr-defined]
406        func.api_allow_impersonation = allow_impersonation  # type: ignore[attr-defined]
407        func.api_alias = alias  # type: ignore[attr-defined]
408        return func
409
410    return decorate
411
412
413def _is_type_hint(value_type: Any) -> bool:
414    """
415    Check if a type annotation is or contains type[X].
416
417    Handles both ``type[X]`` and ``type[X] | None``.
418    """
419    origin = get_origin(value_type)
420    if origin is type:
421        return True
422    if origin is Union or origin is UnionType:
423        return any(get_origin(arg) is type for arg in get_args(value_type))
424    return False
425
426
427def parse_arguments(
428    func_sig: inspect.Signature,
429    func_types: dict[str, Any],
430    args: dict[str, Any] | None,
431    strict: bool = False,
432) -> dict[str, Any]:
433    """Parse (and convert) incoming arguments to correct types."""
434    if args is None:
435        args = {}
436    final_args = {}
437    # ignore extra args if not strict
438    if strict:
439        for key, value in args.items():
440            if key not in func_sig.parameters:
441                raise KeyError(f"Invalid parameter: '{key}'")
442    # parse arguments to correct type
443    for name, param in func_sig.parameters.items():
444        value_type = func_types[name]
445        # Skip type[X] parameters — these are for static type checking only
446        # and must not be resolved from API input.
447        if _is_type_hint(value_type):
448            continue
449        value = args.get(name)
450        default = MISSING if param.default is inspect.Parameter.empty else param.default
451        try:
452            final_args[name] = parse_value(name, value, value_type, default)
453        except TypeError:
454            # retry one more time with allow_value_convert=True
455            final_args[name] = parse_value(
456                name, value, value_type, default, allow_value_convert=True
457            )
458    return final_args
459
460
461def parse_utc_timestamp(datetime_string: str) -> datetime:
462    """Parse datetime from string."""
463    return datetime.fromisoformat(datetime_string)
464
465
466def parse_value(
467    name: str,
468    value: Any,
469    value_type: Any,
470    default: Any = MISSING,
471    allow_value_convert: bool = False,
472) -> Any:
473    """Try to parse a value from raw (json) data and type annotations."""
474    # Resolve string type hints early for proper handling
475    if isinstance(value_type, str):
476        value_type = _resolve_string_type(value_type)
477        # If still a string after resolution, return value as-is
478        if isinstance(value_type, str):
479            LOGGER.debug("Unknown string type hint: %s, returning value as-is", value_type)
480            return value
481
482    if isinstance(value, dict) and hasattr(value_type, "from_dict"):
483        # Only validate media_type for actual MediaItem subclasses, not for other classes
484        # like StreamDetails that have a media_type field for a different purpose
485        if (
486            "media_type" in value
487            and value_type.__name__ != "ItemMapping"
488            and issubclass(value_type, MediaItem)
489            and value["media_type"] != value_type.media_type
490        ):
491            msg = "Invalid MediaType"
492            raise ValueError(msg)
493        return value_type.from_dict(value)
494
495    if value is None and not isinstance(default, type(MISSING)):
496        return default
497    if value is None and value_type is NoneType:
498        return None
499    origin = get_origin(value_type)
500    if origin is tuple and (subtypes := get_args(value_type)) and subtypes[-1] is not Ellipsis:
501        return _parse_fixed_length_tuple(name, value, value_type, subtypes, allow_value_convert)
502    if origin in (tuple, list, set, frozenset, Sequence, Iterable):
503        return _parse_sequence(name, value, value_type, origin, allow_value_convert)
504    if origin is dict:
505        return _parse_dict(value, value_type, allow_value_convert)
506    if origin is Union or origin is UnionType:
507        return _parse_union(name, value, value_type, allow_value_convert)
508    if origin is type:
509        # type[X] parameters are skipped in parse_arguments so this branch
510        # should not be reachable from API input. Reject as a safeguard.
511        msg = f"Cannot resolve type from string: {value!r}"
512        raise ValueError(msg)
513    if value_type is Any:
514        return value
515    if value is None and value_type is not NoneType:
516        msg = f"`{name}` of type `{value_type}` is required."
517        raise KeyError(msg)
518
519    try:
520        if issubclass(value_type, Enum):
521            return value_type(value)
522        if issubclass(value_type, datetime):
523            assert isinstance(value, str)  # for type checking
524            return parse_utc_timestamp(value)
525    except TypeError:
526        # happens if value_type is not a class
527        pass
528
529    if allow_value_convert:
530        value = _convert_common_value(value, value_type)
531
532    if not isinstance(value, value_type):
533        # all options failed, raise exception
534        msg = (
535            f"Value {value} of type {type(value)} is invalid for {name}, "
536            f"expected value of type {value_type}"
537        )
538        raise TypeError(msg)
539    return value
540
541
542def _parse_fixed_length_tuple(
543    name: str,
544    value: Any,
545    value_type: Any,
546    subtypes: tuple[Any, ...],
547    allow_value_convert: bool,
548) -> tuple[Any, ...]:
549    """
550    Parse a value against a fixed-length tuple annotation.
551
552    :param name: Name of the value, used in error messages.
553    :param value: The raw (json) value to parse.
554    :param value_type: The tuple annotation to parse against.
555    :param subtypes: The type arguments of the tuple annotation, one per position.
556    :param allow_value_convert: Whether conversion of common type mistakes is allowed.
557    """
558    # a fixed-length tuple annotates every position separately, so each member is
559    # parsed against its own type and kept, including the members that are None
560    if len(value) != len(subtypes):
561        msg = (
562            f"Value {value} of type {type(value)} is invalid for {name}, "
563            f"expected value of type {value_type}"
564        )
565        raise TypeError(msg)
566    return tuple(
567        parse_value(f"{name}[{index}]", subvalue, subtype, allow_value_convert=allow_value_convert)
568        for index, (subvalue, subtype) in enumerate(zip(value, subtypes, strict=True))
569    )
570
571
572def _parse_sequence(
573    name: str,
574    value: Any,
575    value_type: Any,
576    origin: Any,
577    allow_value_convert: bool,
578) -> Any:
579    """
580    Parse a value against a homogeneous sequence annotation.
581
582    :param name: Name of the value, used in error messages.
583    :param value: The raw (json) value to parse.
584    :param value_type: The sequence annotation to parse against.
585    :param origin: The unsubscripted origin of the annotation.
586    :param allow_value_convert: Whether conversion of common type mistakes is allowed.
587    """
588    subtypes = get_args(value_type)
589    # a None member is only kept when the element type admits it, so a plain annotation
590    # like list[str] still shrugs off the nulls a client may have sent along
591    keep_none = bool(subtypes) and (
592        subtypes[0] is NoneType
593        or (get_origin(subtypes[0]) in (Union, UnionType) and NoneType in get_args(subtypes[0]))
594    )
595    # For abstract types like Sequence and Iterable, use list as the concrete type
596    concrete_type = list if origin in (Sequence, Iterable) else origin
597    return concrete_type(
598        parse_value(name, subvalue, subtypes[0], allow_value_convert=allow_value_convert)
599        for subvalue in value
600        if subvalue is not None or keep_none
601    )
602
603
604def _parse_dict(
605    value: Any,
606    value_type: Any,
607    allow_value_convert: bool,
608) -> dict[Any, Any]:
609    """
610    Parse a value against a dict annotation.
611
612    :param value: The raw (json) value to parse.
613    :param value_type: The dict annotation to parse against.
614    :param allow_value_convert: Whether conversion of common type mistakes is allowed.
615    """
616    subkey_type = get_args(value_type)[0]
617    subvalue_type = get_args(value_type)[1]
618    return {
619        parse_value(subkey, subkey, subkey_type): parse_value(
620            f"{subkey}.value", subvalue, subvalue_type, allow_value_convert=allow_value_convert
621        )
622        for subkey, subvalue in value.items()
623    }
624
625
626def _parse_union(
627    name: str,
628    value: Any,
629    value_type: Any,
630    allow_value_convert: bool,
631) -> Any:
632    """
633    Parse a value against a union annotation.
634
635    :param name: Name of the value, used in error messages.
636    :param value: The raw (json) value to parse.
637    :param value_type: The union annotation to parse against.
638    :param allow_value_convert: Whether conversion of common type mistakes is allowed.
639    """
640    sub_value_types = get_args(value_type)
641    if value is None and NoneType in sub_value_types:
642        # an optional annotation with no value needs no further parsing
643        return None
644    for sub_arg_type in sub_value_types:
645        # try them all until one succeeds
646        try:
647            return parse_value(name, value, sub_arg_type, allow_value_convert=allow_value_convert)
648        except KeyError, TypeError, ValueError, MissingField:
649            pass
650    # if we get to this point, all possibilities failed
651    # find out if we should raise or log this
652    err = (
653        f"Value {value} of type {type(value)} is invalid for {name}, "
654        f"expected value of type {value_type}"
655    )
656    if NoneType not in sub_value_types:
657        # raise exception, we have no idea how to handle this value
658        raise TypeError(err)
659    # failed to parse the (sub) value but None allowed, log only
660    logging.getLogger(__name__).warning(err)
661    return None
662
663
664def _convert_common_value(value: Any, value_type: Any) -> Any:
665    """
666    Convert common type mistakes in raw (json) data, or return the value untouched.
667
668    :param value: The raw (json) value to convert.
669    :param value_type: The type to convert the value to.
670    """
671    if value_type is float and isinstance(value, int):
672        return float(value)
673    if value_type is int and isinstance(value, float):
674        return int(value)
675    if value_type is int and isinstance(value, str) and value.isnumeric():
676        return int(value)
677    if value_type is float and isinstance(value, str) and value.isnumeric():
678        return float(value)
679    if value_type is bool and isinstance(value, str | int):
680        return try_parse_bool(value)
681    return value
682