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