music-assistant-server
32.7 KB•PY
flows.py
32.7 KB • 749 lines • python
1"""Setup flow engine: registry and API commands for interactive provider/player setup."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import time
8from dataclasses import dataclass
9from functools import partial
10from typing import TYPE_CHECKING, Any
11from uuid import uuid4
12
13from music_assistant_models.auth import Scope
14from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
15from music_assistant_models.enums import (
16 ConfigEntryType,
17 EventType,
18 FlowStepType,
19 MediaType,
20 ProviderStage,
21)
22from music_assistant_models.errors import (
23 ActionUnavailable,
24 InsufficientPermissions,
25 SetupFailedError,
26)
27from music_assistant_models.setup_flow import SetupFlowStep
28
29from music_assistant.constants import CONF_PLAYERS, CONF_PROVIDERS
30from music_assistant.controllers.config.helpers import _AUTH_ERROR_CODES
31from music_assistant.helpers.api import api_command
32from music_assistant.helpers.util import import_module_in_thread, load_provider_module
33from music_assistant.models.player import Player
34from music_assistant.models.setup_flow import (
35 AbortFlow,
36 FlowReason,
37 SetupFlowContext,
38 SetupFlowError,
39 SetupSession,
40 StepExpiredError,
41)
42
43if TYPE_CHECKING:
44 from collections.abc import Awaitable, Callable
45
46 from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
47 from music_assistant_models.provider import ProviderManifest
48
49 from music_assistant import MusicAssistant
50
51LOGGER = logging.getLogger(__name__)
52
53# a flow with no client interaction or step changes for this long is garbage collected
54IDLE_FLOW_TTL = 15 * 60
55FLOW_SWEEP_INTERVAL = 60
56# how long start/submit wait for the flow coroutine to produce the (next) step;
57# generous because finish() may install requirements and load the provider
58NEXT_STEP_TIMEOUT = 120
59# bound on waiting for a cancelled flow's cleanup (author finally-blocks)
60FLOW_ABORT_CLEANUP_TIMEOUT = 10
61
62
63@dataclass
64class ActiveSetupFlow:
65 """Registry record for a running setup flow."""
66
67 session: SetupSession
68 # target_key: identifies what the flow is (re)configuring; one flow per target
69 target_key: str
70 # required_scope: the scope the starting command required; re-checked on
71 # every flows/* continuation command
72 required_scope: Scope
73 task: asyncio.Task[None] | None = None
74
75
76class SetupFlowMixin:
77 """Mixin providing the setup flow engine for the ConfigController."""
78
79 # registry of running flows, keyed by flow_id (lazily created per instance)
80 _flows: dict[str, ActiveSetupFlow] | None = None
81 _flow_sweep_handle: asyncio.TimerHandle | None = None
82 # required scopes of recently finished flows: terminal steps can publish after
83 # the registry pop (cancel-driven aborts), and the event-scope filter still
84 # needs to resolve them; bounded FIFO
85 _finished_flow_scopes: dict[str, Scope] | None = None
86
87 # Type hints for attributes/methods provided by the class this mixin is used with
88 if TYPE_CHECKING:
89 mass: MusicAssistant
90
91 @property
92 def onboard_done(self) -> bool: ... # noqa: D102
93
94 def get(self, key: str, default: Any = None) -> Any: ... # noqa: D102
95
96 def set(self, key: str, value: Any, immediate: bool = False) -> None: ... # noqa: D102
97
98 def encrypt_string(self, str_value: str) -> str: ... # noqa: D102
99
100 def decrypt_string(self, encrypted_str: str) -> str: ... # noqa: D102
101
102 async def get_provider_configs( # noqa: D102
103 self, provider_type: Any = None, provider_domain: str | None = None
104 ) -> list[ProviderConfig]: ...
105
106 async def get_provider_config(self, instance_id: str) -> ProviderConfig: ... # noqa: D102
107
108 def update_provider_last_error(self, instance_id: str, error: Any) -> None: ... # noqa: D102
109
110 async def get_player_config(self, player_id: str) -> Any: ... # noqa: D102
111
112 async def _create_provider_instance(
113 self,
114 provider_domain: str,
115 values: dict[str, ConfigValueType],
116 setup_data: dict[str, Any] | None = None,
117 ) -> ProviderConfig: ...
118
119 @api_command("config/providers/setup", required_scope=Scope.CONFIG_PROVIDERS_WRITE)
120 async def setup_provider(self, provider_domain: str) -> SetupFlowStep:
121 """
122 Start the setup flow to add a new instance of the given provider.
123
124 For a provider without a setup flow (no user input needed) the instance is
125 created right away and a FINISH step is returned, so the add-provider path
126 is uniform for all providers.
127
128 :param provider_domain: Domain of the provider to add an instance of.
129 """
130 for manifest in self.mass.get_provider_manifests():
131 if manifest.domain == provider_domain:
132 break
133 else:
134 msg = f"Unknown provider domain: {provider_domain}"
135 raise KeyError(msg)
136 owner = f"provider.{provider_domain}"
137 # fail fast on conditions that would otherwise only surface at save
138 if manifest.stage == ProviderStage.DEPRECATED:
139 # a retired provider can never be set up again; its own strings explain
140 # what to use instead
141 return self._synthesized_step(FlowStepType.ABORT, owner, reason="provider_retired")
142 existing = await self.get_provider_configs(provider_domain=provider_domain)
143 if existing and not manifest.multi_instance:
144 return self._synthesized_step(FlowStepType.ABORT, owner, reason="already_configured")
145 if manifest.depends_on:
146 dep_configs = await self.get_provider_configs(provider_domain=manifest.depends_on)
147 if not any(dep_conf.enabled for dep_conf in dep_configs):
148 return self._synthesized_step(
149 FlowStepType.ABORT, owner, reason="missing_dependency"
150 )
151 flow_module = await self._get_setup_flow_module(manifest)
152 if flow_module is None:
153 # zero-input provider: create the instance immediately and report it
154 # as an (already) finished flow
155 config = await self._create_provider_instance(provider_domain, {})
156 return self._synthesized_step(
157 FlowStepType.FINISH,
158 owner,
159 step_id=self._provider_finish_step_id(config.instance_id),
160 result={"instance_id": config.instance_id},
161 )
162 context = SetupFlowContext(kind="setup", reason="user", domain=provider_domain)
163 return await self._start_flow(
164 flow_coro=flow_module.run_setup,
165 context=context,
166 target_key=f"provider_setup:{provider_domain}",
167 required_scope=Scope.CONFIG_PROVIDERS_WRITE,
168 finish_handler=self._finish_provider_setup,
169 )
170
171 @api_command("config/providers/reconfigure", required_scope=Scope.CONFIG_PROVIDERS_WRITE)
172 async def reconfigure_provider(self, instance_id: str) -> SetupFlowStep:
173 """
174 Start the reconfigure flow on an existing provider instance (covers reauth).
175
176 :param instance_id: The provider instance to reconfigure.
177 """
178 raw_conf = self.get(f"{CONF_PROVIDERS}/{instance_id}")
179 if not raw_conf:
180 msg = f"No config found for provider id {instance_id}"
181 raise KeyError(msg)
182 domain: str = raw_conf["domain"]
183 manifest = self.mass.get_provider_manifest(domain)
184 owner = f"provider.{domain}"
185 flow_module = await self._get_setup_flow_module(manifest)
186 if flow_module is None:
187 # flow-less providers have nothing to reconfigure;
188 # their failures are environmental (reload/retry covers them)
189 return self._synthesized_step(FlowStepType.ABORT, owner, reason="nothing_to_configure")
190 context = SetupFlowContext(
191 kind="reconfigure",
192 reason=self._reconfigure_reason(raw_conf.get("last_error")),
193 domain=domain,
194 instance_id=instance_id,
195 setup_data=self._decrypt_values(raw_conf.get("setup_data") or {}),
196 values=self._decrypt_values(raw_conf.get("values") or {}),
197 )
198 return await self._start_flow(
199 flow_coro=flow_module.run_setup,
200 context=context,
201 target_key=f"provider_reconfigure:{instance_id}",
202 required_scope=Scope.CONFIG_PROVIDERS_WRITE,
203 finish_handler=self._finish_provider_reconfigure,
204 )
205
206 @api_command("config/players/setup", required_scope=Scope.CONFIG_PLAYERS_WRITE)
207 async def setup_player(self, player_id: str) -> SetupFlowStep:
208 """
209 Start the setup flow for a player (e.g. pairing).
210
211 A player that itself needs no setup but wraps protocol child player(s) that do
212 (universal players, or native players wrapping protocol children) delegates to
213 the child's setup flow: to the single child that needs setup directly, or - when
214 more than one does - via a form that lets the user pick which child to set up.
215 The child's flow persists to the child's own config.
216
217 Also serves on-demand re-runs: when nothing needs setup (anymore), delegation
218 falls back to any child that merely has a flow, so a step the user skipped
219 earlier - an optional pairing, say - remains reachable.
220
221 :param player_id: The player to set up.
222 """
223 # deliberately no raise_unavailable: a player that needs setup is serialized
224 # as unavailable, and that is exactly the player this command targets
225 player = self.mass.players.get_player(player_id)
226 if player is None:
227 msg = f"Player {player_id} not found"
228 raise KeyError(msg)
229 owner = f"provider.{player.provider.domain}"
230 target_key = f"player_setup:{player_id}"
231 if player.implements_setup_flow:
232 # the player implements its own setup flow: run it directly
233 return await self._start_flow(
234 flow_coro=player.run_setup_flow,
235 context=self._player_flow_context(player),
236 target_key=target_key,
237 required_scope=Scope.CONFIG_PLAYERS_WRITE,
238 finish_handler=self._finish_player_setup,
239 )
240 # no direct setup: delegate to protocol child player(s), preferring the ones
241 # that actually need setup and falling back to any that can re-run their flow
242 children = self._protocol_children_with_setup_flow(player, needing_only=True)
243 if not children:
244 children = self._protocol_children_with_setup_flow(player, needing_only=False)
245 if len(children) == 1:
246 child = children[0]
247 return await self._start_flow(
248 flow_coro=child.run_setup_flow,
249 context=self._player_flow_context(child),
250 # key on the child: a direct setup of the child must replace this flow
251 target_key=f"player_setup:{child.player_id}",
252 required_scope=Scope.CONFIG_PLAYERS_WRITE,
253 finish_handler=self._finish_player_setup,
254 )
255 if children:
256 return await self._start_flow(
257 flow_coro=partial(self._run_child_selection_flow, children),
258 context=self._player_flow_context(player),
259 target_key=target_key,
260 required_scope=Scope.CONFIG_PLAYERS_WRITE,
261 finish_handler=self._finish_player_setup,
262 )
263 # nothing on this player (or its children) to configure
264 return self._synthesized_step(FlowStepType.ABORT, owner, reason="nothing_to_configure")
265
266 @api_command("config/flows/submit")
267 async def submit_setup_flow(
268 self, flow_id: str, values: dict[str, ConfigValueType]
269 ) -> SetupFlowStep:
270 """
271 Submit the user's values for the flow's pending FORM step.
272
273 Returns the flow's next step, or the same FORM step (with per-field errors
274 set) when validation failed.
275
276 :param flow_id: The id of the running flow.
277 :param values: The raw values for the form's config entries.
278 """
279 flow = self._get_flow(flow_id)
280 self._check_flow_permission(flow)
281 if (error_step := flow.session.handle_submit(values)) is not None:
282 return error_step
283 # wait (bounded) for the coroutine to produce the next step
284 submitted_step = flow.session.current_step
285 await flow.session.wait_for_step_change(NEXT_STEP_TIMEOUT)
286 step = flow.session.current_step
287 assert step is not None # an accepted submit implies a published FORM step
288 if step is submitted_step:
289 # rare: the coroutine is still working on the next step. The submitted
290 # form's input future is already consumed, so re-serving the form would
291 # invite a doomed resubmit - publish a progress step (so flows/get agrees)
292 # and let the coroutine's next publish deliver the real step
293 flow.session.progress("working")
294 step = flow.session.current_step
295 assert step is not None
296 return step
297
298 @api_command("config/flows/get")
299 async def get_setup_flow(self, flow_id: str) -> SetupFlowStep:
300 """
301 Return the current step of a running flow (idempotent re-render, never advances).
302
303 :param flow_id: The id of the running flow.
304 """
305 flow = self._get_flow(flow_id)
306 self._check_flow_permission(flow)
307 flow.session.last_activity = time.monotonic()
308 if (step := flow.session.current_step) is None:
309 raise ActionUnavailable("The setup flow has not produced a step yet")
310 return step
311
312 @api_command("config/flows/abort")
313 async def abort_setup_flow(self, flow_id: str) -> None:
314 """
315 Abort a running flow (user cancelled).
316
317 :param flow_id: The id of the running flow.
318 """
319 flow = self._get_flow(flow_id)
320 self._check_flow_permission(flow)
321 await self._abort_flow(flow, reason="aborted")
322
323 def get_setup_flow_required_scope(self, flow_id: str) -> Scope | None:
324 """
325 Return the scope required to receive/interact with the given setup flow.
326
327 Also resolves recently finished flows (their terminal step can publish
328 just after the registry pop). Returns None when the flow is unknown.
329
330 :param flow_id: The id of the flow.
331 """
332 if flow := self._setup_flows.get(flow_id):
333 return flow.required_scope
334 if self._finished_flow_scopes:
335 return self._finished_flow_scopes.get(flow_id)
336 return None
337
338 def _pop_flow(self, flow: ActiveSetupFlow) -> None:
339 """Remove a flow from the registry, retaining its scope for late events."""
340 self._setup_flows.pop(flow.session.flow_id, None)
341 if self._finished_flow_scopes is None:
342 self._finished_flow_scopes = {}
343 finished = self._finished_flow_scopes
344 finished[flow.session.flow_id] = flow.required_scope
345 while len(finished) > 64:
346 finished.pop(next(iter(finished)))
347
348 async def _start_flow(
349 self,
350 *,
351 flow_coro: Callable[[SetupSession], Awaitable[Any]],
352 context: SetupFlowContext,
353 target_key: str,
354 required_scope: Scope,
355 finish_handler: Callable[
356 [SetupSession, dict[str, ConfigValueType]], Awaitable[dict[str, str]]
357 ],
358 ) -> SetupFlowStep:
359 """Register and start a new flow, returning its first published step."""
360 # one flow per target: starting anew replaces (aborts) a lingering previous flow.
361 # re-scan after every await: the abort yields, so a concurrent start for the same
362 # target may have registered a new flow in the meantime
363 while existing_flow := next(
364 (f for f in self._setup_flows.values() if f.target_key == target_key), None
365 ):
366 await self._abort_flow(existing_flow, reason="replaced")
367 flow_id = uuid4().hex
368 session = SetupSession(self.mass, flow_id, context, finish_handler)
369 flow = ActiveSetupFlow(
370 session=session, target_key=target_key, required_scope=required_scope
371 )
372 self._setup_flows[flow_id] = flow
373 LOGGER.debug("Starting setup flow %s for %s", flow_id, target_key)
374 flow.task = self.mass.create_task(self._run_flow(flow, flow_coro))
375 self._schedule_flow_sweep()
376 if session.current_step is None:
377 await session.wait_for_step_change(NEXT_STEP_TIMEOUT)
378 if (step := session.current_step) is None:
379 await self._abort_flow(flow, reason="internal_error")
380 raise SetupFailedError("Setup flow did not produce a first step")
381 return step
382
383 async def _run_flow(
384 self, flow: ActiveSetupFlow, flow_coro: Callable[[SetupSession], Awaitable[Any]]
385 ) -> None:
386 """Drive the flow coroutine and convert its outcome into a terminal step."""
387 session = flow.session
388 try:
389 await flow_coro(session)
390 except AbortFlow as err:
391 session.publish_abort(err.reason)
392 except StepExpiredError:
393 session.publish_abort("timed_out")
394 except SetupFlowError as err:
395 # the author did not catch a finish failure: end with the failure message
396 session.publish_abort(str(err) or "internal_error")
397 except asyncio.CancelledError:
398 # abort/replace/shutdown: the author's cleanup (finally blocks) has run;
399 # the canceller publishes the ABORT step. Never swallow the cancellation.
400 raise
401 except Exception:
402 LOGGER.exception("Unhandled error in setup flow for %s", session.context.domain)
403 session.publish_abort("internal_error")
404 else:
405 if not session.finished:
406 LOGGER.error(
407 "Setup flow for %s returned without calling finish()", session.context.domain
408 )
409 session.publish_abort("internal_error")
410 finally:
411 LOGGER.debug("Setup flow %s ended", session.flow_id)
412 session.close()
413 self._pop_flow(flow)
414
415 async def _abort_flow(self, flow: ActiveSetupFlow, reason: str) -> None:
416 """
417 Abort the given flow with the given reason and clean it up.
418
419 Cancelling raises CancelledError inside the flow coroutine so the author's
420 cleanup (try/finally around pairing sessions etc.) runs before the terminal
421 ABORT step goes out.
422 """
423 if flow.task is not None and not flow.task.done():
424 flow.task.cancel()
425 # wait() shields us from the task's CancelledError without
426 # masking a cancellation of the caller itself; the timeout keeps a
427 # wedged author cleanup (e.g. a hanging pairing teardown) from
428 # stalling the abort and any replacement flow indefinitely
429 _, pending = await asyncio.wait([flow.task], timeout=FLOW_ABORT_CLEANUP_TIMEOUT)
430 if pending:
431 LOGGER.warning(
432 "Setup flow for %s did not clean up within %ss after cancellation",
433 flow.session.context.domain,
434 FLOW_ABORT_CLEANUP_TIMEOUT,
435 )
436 # the wedged task never reaches _run_flow's finally: close the
437 # session here so the unauthenticated callback route is dropped
438 flow.session.close()
439 self._pop_flow(flow)
440 current_step = flow.session.current_step
441 if current_step is None or current_step.type not in (
442 FlowStepType.FINISH,
443 FlowStepType.ABORT,
444 ):
445 flow.session.publish_abort(reason)
446
447 async def _finish_provider_setup(
448 self, session: SetupSession, values: dict[str, ConfigValueType]
449 ) -> dict[str, str]:
450 """Finish handler for provider setup flows: create, persist and load the instance."""
451 try:
452 config = await self._create_provider_instance(
453 session.context.domain, {}, setup_data=self._encrypt_values(values)
454 )
455 except Exception as err:
456 raise SetupFlowError(
457 str(err) or err.__class__.__name__,
458 translation_key=getattr(err, "translation_key", None),
459 ) from err
460 session.finish_step_id = self._provider_finish_step_id(config.instance_id)
461 return {"instance_id": config.instance_id}
462
463 async def _finish_provider_reconfigure(
464 self, session: SetupSession, values: dict[str, ConfigValueType]
465 ) -> dict[str, str]:
466 """Finish handler for provider reconfigure flows: merge setup_data and reload."""
467 instance_id = session.context.instance_id
468 assert instance_id is not None # always set for reconfigure flows
469 conf_key = f"{CONF_PROVIDERS}/{instance_id}"
470 raw_conf = self.get(conf_key)
471 if not raw_conf:
472 raise SetupFlowError(f"Provider {instance_id} no longer exists")
473 snapshot = dict(raw_conf.get("setup_data") or {})
474 self.set(f"{conf_key}/setup_data", {**snapshot, **self._encrypt_values(values)})
475 try:
476 config = await self.get_provider_config(instance_id)
477 await self.mass.load_provider_config(config)
478 except asyncio.CancelledError:
479 self.set(f"{conf_key}/setup_data", snapshot)
480 raise
481 except Exception as err:
482 # reloading with the new values failed: restore the previous setup_data
483 self.set(f"{conf_key}/setup_data", snapshot)
484 raise SetupFlowError(
485 str(err) or err.__class__.__name__,
486 translation_key=getattr(err, "translation_key", None),
487 ) from err
488 self.update_provider_last_error(instance_id, None)
489 return {"instance_id": instance_id}
490
491 async def _finish_player_setup(
492 self, session: SetupSession, values: dict[str, ConfigValueType]
493 ) -> dict[str, str]:
494 """Finish handler for player setup flows: persist and apply the collected setup data."""
495 player_id = session.context.player_id
496 assert player_id is not None # always set for player flows
497 conf_key = f"{CONF_PLAYERS}/{player_id}"
498 raw_conf = self.get(conf_key)
499 if not raw_conf:
500 raise SetupFlowError(f"No config found for player {player_id}")
501 snapshot = dict(raw_conf.get("setup_data") or {})
502 self.set(f"{conf_key}/setup_data", {**snapshot, **self._encrypt_values(values)})
503 try:
504 config = await self.get_player_config(player_id)
505 changed_keys = {f"setup_data/{key}" for key in values}
506 await self.mass.players.on_player_config_change(config, changed_keys)
507 except asyncio.CancelledError:
508 self.set(f"{conf_key}/setup_data", snapshot)
509 raise
510 except Exception as err:
511 # reading back or applying the updated config failed: restore the previous setup_data
512 self.set(f"{conf_key}/setup_data", snapshot)
513 raise SetupFlowError(
514 str(err) or err.__class__.__name__,
515 translation_key=getattr(err, "translation_key", None),
516 ) from err
517 self.mass.signal_event(EventType.PLAYER_CONFIG_UPDATED, object_id=player_id, data=config)
518 return {"player_id": player_id}
519
520 def _player_flow_context(self, player: Player) -> SetupFlowContext:
521 """Build the setup flow context (with decrypted prefill) for the given player."""
522 raw_conf = self.get(f"{CONF_PLAYERS}/{player.player_id}") or {}
523 return SetupFlowContext(
524 kind="setup",
525 reason="user",
526 domain=player.provider.domain,
527 instance_id=player.provider.instance_id,
528 player_id=player.player_id,
529 setup_data=self._decrypt_values(raw_conf.get("setup_data") or {}),
530 values=self._decrypt_values(raw_conf.get("values") or {}),
531 )
532
533 def _protocol_children_with_setup_flow(
534 self, player: Player, *, needing_only: bool
535 ) -> list[Player]:
536 """
537 Return the player's protocol child players whose setup flow is available.
538
539 Covers the wrapper case: a universal player, or a native player wrapping
540 protocol children, whose own setup is a no-op but whose linked protocol
541 outputs still require pairing/credentials.
542
543 :param player: The (wrapper) player whose protocol children to inspect.
544 :param needing_only: Only return children that currently need setup.
545 """
546 children: list[Player] = []
547 seen: set[str] = set()
548 for output_protocol in player.output_protocols:
549 child_id = output_protocol.output_protocol_id
550 if output_protocol.is_native or child_id in seen:
551 continue
552 seen.add(child_id)
553 child = self.mass.players.get_player(child_id)
554 if child is None or not child.has_setup_flow:
555 continue
556 if needing_only and not child.needs_setup:
557 continue
558 children.append(child)
559 return children
560
561 async def _run_child_selection_flow(
562 self, children: list[Player], session: SetupSession
563 ) -> None:
564 """
565 Run the wrapper flow that lets the user pick which protocol child to set up.
566
567 The selection form is owned by the parent; once a child is picked the session is
568 re-pointed at that child so its flow's steps localize and its ``finish()`` persists
569 under the child's own config.
570 """
571 options = [
572 ConfigValueOption(
573 value=child.player_id, title=f"{child.display_name} ({child.provider.name})"
574 )
575 for child in children
576 ]
577 values = await session.form(
578 [
579 ConfigEntry(
580 key="child",
581 type=ConfigEntryType.STRING,
582 required=True,
583 default_value=children[0].player_id,
584 options=options,
585 )
586 ],
587 step_id="select_child",
588 )
589 child_id = str(values["child"])
590 child = next((candidate for candidate in children if candidate.player_id == child_id), None)
591 if child is None:
592 raise AbortFlow("nothing_to_configure")
593 child_raw = self.get(f"{CONF_PLAYERS}/{child_id}") or {}
594 session.retarget(
595 domain=child.provider.domain,
596 instance_id=child.provider.instance_id,
597 player_id=child_id,
598 setup_data=self._decrypt_values(child_raw.get("setup_data") or {}),
599 values=self._decrypt_values(child_raw.get("values") or {}),
600 )
601 await child.run_setup_flow(session)
602
603 async def _get_setup_flow_module(self, manifest: ProviderManifest) -> Any | None:
604 """
605 Import (lazily) the provider's setup_flow module, or None when it has none.
606
607 A provider without a setup_flow module needs no setup input at all.
608 """
609 # ensure the provider's requirements are installed and its package imports
610 # cleanly first: the setup_flow submodule may rely on those requirements
611 await load_provider_module(manifest.domain, manifest.requirements)
612 module_path = f"music_assistant.providers.{manifest.domain}.setup_flow"
613 try:
614 return await import_module_in_thread(module_path)
615 except ModuleNotFoundError as err:
616 if err.name == module_path:
617 # the provider ships no setup_flow module: it needs no setup input
618 return None
619 # an import *inside* setup_flow.py failed: an actual bug, surface it
620 raise
621
622 @property
623 def _setup_flows(self) -> dict[str, ActiveSetupFlow]:
624 """Return the registry of running flows (created lazily)."""
625 if self._flows is None:
626 self._flows = {}
627 return self._flows
628
629 def _get_flow(self, flow_id: str) -> ActiveSetupFlow:
630 """Return the running flow for the given id."""
631 if flow := self._setup_flows.get(flow_id):
632 return flow
633 msg = f"Unknown (or finished) setup flow: {flow_id}"
634 raise KeyError(msg)
635
636 def _check_flow_permission(self, flow: ActiveSetupFlow) -> None:
637 """Verify the calling user holds the scope the flow's start command required."""
638 # imported here: the webserver helpers pull in the full auth stack,
639 # which must not be imported with the config controller at startup
640 from music_assistant.controllers.webserver.helpers.auth_middleware import ( # noqa: PLC0415
641 get_current_user,
642 has_scope,
643 )
644
645 user = get_current_user()
646 # no user context means an internal (server-side) caller, which is trusted
647 if user is not None and not has_scope(user, flow.required_scope):
648 raise InsufficientPermissions(
649 f"This action requires the {flow.required_scope.value} scope"
650 )
651
652 def _synthesized_step(
653 self,
654 step_type: FlowStepType,
655 translation_owner: str,
656 *,
657 step_id: str | None = None,
658 result: dict[str, str] | None = None,
659 reason: str | None = None,
660 ) -> SetupFlowStep:
661 """
662 Return a terminal step for a flow that ended before a session was needed.
663
664 :param step_type: The terminal step type (FINISH or ABORT).
665 :param translation_owner: The namespace the step's strings resolve under.
666 :param step_id: Slug to serve the step's strings under; defaults to the one
667 implied by the step type.
668 :param result: Reference to the created/updated object (FINISH).
669 :param reason: Slug describing why the flow was aborted (ABORT).
670 """
671 default_step_id = "finish" if step_type == FlowStepType.FINISH else "abort"
672 return SetupFlowStep(
673 flow_id=uuid4().hex,
674 step_id=step_id or default_step_id,
675 type=step_type,
676 result=result,
677 reason=reason,
678 translation_owner=translation_owner,
679 )
680
681 def _provider_finish_step_id(self, instance_id: str) -> str:
682 """
683 Return the i18n slug of the FINISH step for a newly set up provider instance.
684
685 :param instance_id: The provider instance the flow just created.
686 """
687 # a provider that imports a library gets the variant explaining that the first import
688 # runs in the background, so a library that still looks empty right after setup does
689 # not read as a failed setup
690 provider = self.mass.get_provider(instance_id, return_unavailable=True)
691 if provider is not None and any(
692 self.mass.music.library_supported(provider, media_type) for media_type in MediaType.ALL
693 ):
694 return "finish_library_sync"
695 return "finish"
696
697 def _reconfigure_reason(self, last_error: Any) -> FlowReason:
698 """Derive the reconfigure flow reason from the provider's stored last_error."""
699 if not last_error:
700 return "user"
701 # legacy settings may still hold a plain string last_error
702 error_code = last_error.get("error_code") if isinstance(last_error, dict) else None
703 if error_code in _AUTH_ERROR_CODES:
704 return "auth"
705 return "error"
706
707 def _encrypt_values(self, values: dict[str, ConfigValueType]) -> dict[str, Any]:
708 """Return a copy of the values with all string values encrypted (at-rest form)."""
709 return {
710 key: self.encrypt_string(value) if isinstance(value, str) else value
711 for key, value in values.items()
712 }
713
714 def _decrypt_values(self, values: dict[str, Any]) -> dict[str, Any]:
715 """Return a copy of the values with all (encrypted) string values decrypted."""
716 return {
717 key: self.decrypt_string(value) if isinstance(value, str) else value
718 for key, value in values.items()
719 }
720
721 def _schedule_flow_sweep(self) -> None:
722 """Arm the periodic idle-flow sweeper (idempotent)."""
723 if self._flow_sweep_handle is not None:
724 return
725 self._flow_sweep_handle = self.mass.loop.call_later(
726 FLOW_SWEEP_INTERVAL, self._sweep_idle_flows
727 )
728
729 def _sweep_idle_flows(self) -> None:
730 """Abort flows that have been idle for longer than the TTL."""
731 self._flow_sweep_handle = None
732 if self.mass.closing:
733 return
734 now = time.monotonic()
735 for flow in list(self._setup_flows.values()):
736 current_step = flow.session.current_step
737 if (
738 current_step is not None
739 and current_step.expires_at is not None
740 and current_step.expires_at > time.time()
741 ):
742 # the step advertises a (longer) countdown to the user; the step
743 # deadline machinery guarantees the flow terminates on its own
744 continue
745 if now - flow.session.last_activity >= IDLE_FLOW_TTL:
746 self.mass.create_task(self._abort_flow(flow, "timed_out"))
747 if self._setup_flows:
748 self._schedule_flow_sweep()
749