/
/
1"""
2Session model and exceptions for interactive setup flows.
3
4A setup flow is the interactive, multi-step process (credentials, OAuth, pairing, ...)
5that creates or reconfigures a provider or player. A flow is authored as one plain
6coroutine that awaits the user through a SetupSession:
7
8 # music_assistant/providers/<domain>/setup_flow.py
9 async def run_setup(session: SetupSession) -> None:
10 values = await session.form([...])
11 await session.finish(values)
12
13The engine (config controller's SetupFlowMixin) owns the session lifecycle: it stores
14the current step, pushes SETUP_FLOW_UPDATED events, feeds submitted values/callback
15params back into the awaiting coroutine and persists the collected values on finish.
16"""
17
18from __future__ import annotations
19
20import asyncio
21import logging
22import time
23from dataclasses import dataclass, field, replace
24from typing import TYPE_CHECKING, Any, Literal, TypeVar
25
26from aiohttp.web import Request, Response
27from music_assistant_models.config_entries import UI_ONLY, ConfigEntry, ConfigValueType
28from music_assistant_models.enums import ConfigEntryType, EventType, FlowStepType
29from music_assistant_models.errors import ActionUnavailable
30from music_assistant_models.setup_flow import SetupFlowStep
31
32from music_assistant.helpers.json import json_loads
33
34if TYPE_CHECKING:
35 from collections.abc import Awaitable, Callable
36
37 from music_assistant.mass import MusicAssistant
38
39LOGGER = logging.getLogger(__name__)
40
41_T = TypeVar("_T")
42
43FlowKind = Literal["setup", "reconfigure"]
44FlowReason = Literal["user", "auth", "error"]
45
46CALLBACK_RESPONSE_HTML = """
47<html>
48<body onload="window.close();">
49 Authentication completed, you may now close this window.
50</body>
51</html>
52"""
53
54
55class SetupFlowError(Exception):
56 """
57 Error raised into the flow coroutine when session.finish() failed.
58
59 The author may catch it to re-render a form with the error message and retry,
60 or let it propagate so the engine aborts the flow with the failure message.
61 """
62
63 def __init__(self, message: str, translation_key: str | None = None) -> None:
64 """
65 Initialize the error.
66
67 :param message: Human readable (English) description of the failure.
68 :param translation_key: Optional (bare) translation slug of the underlying error,
69 usable by flow authors as a localizable form error.
70 """
71 super().__init__(message)
72 self.translation_key = translation_key
73
74
75class StepExpiredError(Exception):
76 """
77 Raised into the flow coroutine when the active step's deadline passed.
78
79 The author may catch it to refresh the step (e.g. a new QR code or pairing
80 session) or let it propagate so the engine aborts the flow as timed out.
81 """
82
83
84class AbortFlow(Exception):
85 """
86 Raise anywhere inside a flow coroutine to cleanly abort the flow.
87
88 :param reason: Slug describing why the flow was aborted; resolved from the
89 translations (setup_flow.abort.<reason>) when the ABORT step is served.
90 """
91
92 def __init__(self, reason: str = "aborted") -> None:
93 """Initialize with the abort reason slug."""
94 super().__init__(reason)
95 self.reason = reason
96
97
98@dataclass(kw_only=True)
99class SetupFlowContext:
100 """Static context describing what a running setup flow is (re)configuring."""
101
102 # kind: whether this flow sets up a new target or reconfigures an existing one
103 kind: FlowKind
104 # reason: what triggered the flow (user initiated, auth failure or another error)
105 reason: FlowReason
106 # domain: the provider domain the flow belongs to (for players: the player's provider)
107 domain: str
108 # instance_id: the existing provider instance (reconfigure/player flows)
109 instance_id: str | None = None
110 # player_id: the player being set up (player flows only)
111 player_id: str | None = None
112 # setup_data: decrypted values collected by an earlier flow run, for prefill
113 setup_data: dict[str, Any] = field(default_factory=dict)
114 # values: the target's existing (options) config values, for prefill
115 values: dict[str, ConfigValueType] = field(default_factory=dict)
116
117
118class SetupSession:
119 """
120 Coroutine-facing session handle for a running setup flow.
121
122 Handed to the flow coroutine (``run_setup(session)``); the form/external/progress
123 methods each publish a step to the client(s) and - where applicable - suspend the
124 coroutine until the engine feeds the user's response back in.
125 """
126
127 def __init__(
128 self,
129 mass: MusicAssistant,
130 flow_id: str,
131 context: SetupFlowContext,
132 finish_handler: Callable[
133 [SetupSession, dict[str, ConfigValueType]], Awaitable[dict[str, str]]
134 ],
135 ) -> None:
136 """
137 Initialize the session (done by the flow engine, never by flow authors).
138
139 :param mass: The MusicAssistant instance.
140 :param flow_id: Unique id of this flow.
141 :param context: Static context describing the flow's target.
142 :param finish_handler: Engine callback that persists the collected values
143 and creates/reloads the target when the flow finishes.
144 """
145 self.mass = mass
146 self.flow_id = flow_id
147 self.context = context
148 self.current_step: SetupFlowStep | None = None
149 self.finished = False
150 self.last_activity = time.monotonic()
151 # i18n slug of the terminal FINISH step; the engine swaps in a variant when the
152 # target warrants extra closing copy (e.g. a music provider's initial library import)
153 self.finish_step_id = "finish"
154 self._finish_handler = finish_handler
155 self._translation_owner = f"provider.{context.domain}"
156 self._callback_path = f"/setup_flow/callback/{flow_id}"
157 self._input_future: asyncio.Future[dict[str, ConfigValueType]] | None = None
158 self._callback_future: asyncio.Future[dict[str, str]] | None = None
159 self._step_changed = asyncio.Event()
160 self._unregister_callback_route: Callable[[], None] | None = None
161
162 @property
163 def callback_path(self) -> str:
164 """Return the local callback path that resumes this flow's external step."""
165 self._ensure_callback_route()
166 return self._callback_path
167
168 @property
169 def callback_url(self) -> str:
170 """Return the public callback URL that resumes this flow's external step."""
171 return f"{self.mass.webserver.base_url}{self.callback_path}"
172
173 async def form(
174 self,
175 entries: list[ConfigEntry],
176 step_id: str = "user",
177 errors: dict[str, str] | None = None,
178 last_step: bool | None = None,
179 expires_in: float | None = None,
180 translation_params: list[str] | None = None,
181 ) -> dict[str, ConfigValueType]:
182 """
183 Show a form to the user and wait for the submitted (validated) values.
184
185 :param entries: The config entries that make up the form fields.
186 :param step_id: Stable slug identifying this step (also the i18n key segment).
187 :param errors: Optional field-key (or "base") -> error slug to display.
188 :param last_step: Optional hint that this is the flow's final form.
189 :param expires_in: Optional deadline in seconds; when it passes,
190 StepExpiredError is raised here (and the client countdown runs out).
191 :param translation_params: Optional values for placeholders in the step translations.
192 """
193 step = self._build_step(
194 FlowStepType.FORM,
195 step_id,
196 entries=self._prepare_entries(entries),
197 errors=dict(errors) if errors else {},
198 last_step=last_step,
199 translation_params=translation_params,
200 expires_in=expires_in,
201 )
202 self._input_future = asyncio.get_running_loop().create_future()
203 self._publish_step(step)
204 try:
205 return await self._await_with_deadline(self._input_future, expires_in)
206 finally:
207 self._input_future = None
208
209 async def external(
210 self,
211 url: str,
212 step_id: str = "auth",
213 expires_in: float | None = None,
214 ) -> dict[str, str]:
215 """
216 Send the user to an external URL (e.g. OAuth) and wait for the callback params.
217
218 The flow resumes when the external party (or a bounce page) hits this flow's
219 ``callback_url``; GET query and POST body parameters are merged and returned.
220 As soon as the callback lands a generic progress step replaces the external one,
221 so the client stops asking the user for something they already did.
222
223 :param url: The URL the user must open.
224 :param step_id: Stable slug identifying this step (also the i18n key segment).
225 :param expires_in: Optional deadline in seconds; when it passes,
226 StepExpiredError is raised here (and the client countdown runs out).
227 """
228 self._ensure_callback_route()
229 step = self._build_step(FlowStepType.EXTERNAL, step_id, url=url, expires_in=expires_in)
230 self._callback_future = asyncio.get_running_loop().create_future()
231 self._publish_step(step)
232 try:
233 params = await self._await_with_deadline(self._callback_future, expires_in)
234 finally:
235 self._callback_future = None
236 self.progress("working")
237 return params
238
239 async def external_until(
240 self,
241 awaitable: Awaitable[_T],
242 url: str,
243 step_id: str = "auth",
244 expires_in: float | None = None,
245 translation_params: list[str] | None = None,
246 ) -> _T:
247 """
248 Show an external "Open URL" step that completes when ``awaitable`` resolves.
249
250 Unlike :meth:`external`, which waits for a browser callback, this drives
251 completion from the given awaitable (e.g. a device-code poll) for flows that
252 have no callback to return to. The step renders identically - an Open button for
253 ``url`` plus a waiting spinner - and is dismissed when the awaitable resolves.
254
255 :param awaitable: The work/wait whose completion advances the flow.
256 :param url: The URL the user must open.
257 :param step_id: Stable slug identifying this step (also the i18n key segment).
258 :param expires_in: Optional deadline in seconds; when it passes,
259 StepExpiredError is raised here (and the client countdown runs out).
260 :param translation_params: Optional values for placeholders in the step
261 translations, e.g. a device code the user has to read off the screen.
262 """
263 step = self._build_step(
264 FlowStepType.EXTERNAL,
265 step_id,
266 url=url,
267 expires_in=expires_in,
268 translation_params=translation_params,
269 )
270 self._publish_step(step)
271 return await self._await_with_deadline(awaitable, expires_in)
272
273 def progress(
274 self,
275 step_id: str,
276 text: str | None = None,
277 pct: float | None = None,
278 image: str | None = None,
279 ) -> None:
280 """
281 Publish a display-only progress step (fire-and-forget, re-emittable).
282
283 :param step_id: Stable slug identifying this step (also the i18n key segment).
284 :param text: Optional progress text (slug, resolved from the translations).
285 :param pct: Optional completion fraction between 0 and 1.
286 :param image: Optional data-URI illustration (e.g. a pairing QR code).
287 """
288 step = self._build_step(
289 FlowStepType.PROGRESS, step_id, progress_text=text, progress=pct, image=image
290 )
291 self._publish_step(step)
292
293 async def progress_until(
294 self,
295 awaitable: Awaitable[_T],
296 step_id: str,
297 text: str | None = None,
298 image: str | None = None,
299 expires_in: float | None = None,
300 ) -> _T:
301 """
302 Publish a progress step and wait for the given awaitable, deadline-enforced.
303
304 Display and enforcement come from the single ``expires_in`` declaration: the
305 step's countdown and the engine deadline cannot drift. On the deadline
306 StepExpiredError is raised here and the awaitable is cancelled - this also
307 cancels a pre-existing Task (cancellation propagates through the await);
308 pass ``join_task(task)`` (helpers.util) if a task must survive the deadline.
309
310 :param awaitable: The work/wait to perform while the progress step shows.
311 :param step_id: Stable slug identifying this step (also the i18n key segment).
312 :param text: Optional progress text (slug, resolved from the translations).
313 :param image: Optional data-URI illustration (e.g. a pairing QR code).
314 :param expires_in: Optional deadline in seconds.
315 """
316 step = self._build_step(
317 FlowStepType.PROGRESS, step_id, progress_text=text, image=image, expires_in=expires_in
318 )
319 self._publish_step(step)
320 return await self._await_with_deadline(awaitable, expires_in)
321
322 async def finish(self, values: dict[str, ConfigValueType]) -> dict[str, str]:
323 """
324 Complete the flow: persist the collected values and create/reload the target.
325
326 Raises SetupFlowError when applying the values failed (e.g. the provider
327 does not load with them); the author may catch it and loop back to a form.
328 On success the FINISH step is published and the result reference
329 (e.g. ``{"instance_id": ...}``) is returned.
330
331 :param values: The values to persist as the target's setup_data.
332 """
333 if self.finished:
334 msg = "Setup flow already finished"
335 raise RuntimeError(msg)
336 result = await self._finish_handler(self, values)
337 self.finished = True
338 step = self._build_step(FlowStepType.FINISH, self.finish_step_id, result=result)
339 self._publish_step(step)
340 return result
341
342 def retarget(
343 self,
344 *,
345 domain: str,
346 instance_id: str | None,
347 player_id: str,
348 setup_data: dict[str, Any],
349 values: dict[str, ConfigValueType],
350 ) -> None:
351 """
352 Re-point this running flow at a different (child) player.
353
354 Used by a parent player's wrapper flow to delegate to one of its protocol
355 children: after re-pointing, the child's steps localize under the child's
356 provider namespace and ``finish()`` persists to the child's config. The steps
357 published before the retarget (e.g. a "which device" selection form owned by
358 the parent) keep the parent's namespace.
359
360 :param domain: The child provider's domain.
361 :param instance_id: The child provider instance id.
362 :param player_id: The child player id (the new persist/finish target).
363 :param setup_data: The child's decrypted setup_data, for prefill.
364 :param values: The child's existing config values, for prefill.
365 """
366 self.context = replace(
367 self.context,
368 domain=domain,
369 instance_id=instance_id,
370 player_id=player_id,
371 setup_data=setup_data,
372 values=values,
373 )
374 self._translation_owner = f"provider.{domain}"
375
376 # ------------------------------------------------------------------------------
377 # Engine-facing methods below: called by the SetupFlowMixin, never by flow authors
378 # ------------------------------------------------------------------------------
379
380 def handle_submit(self, values: dict[str, ConfigValueType]) -> SetupFlowStep | None:
381 """
382 Validate submitted form values and hand them to the awaiting flow coroutine.
383
384 Returns the updated FORM step when validation failed (the coroutine is not
385 woken), or None when the values were accepted and the coroutine will resume.
386
387 :param values: The raw values as submitted by the client.
388 """
389 step = self.current_step
390 if (
391 step is None
392 or step.type != FlowStepType.FORM
393 or self._input_future is None
394 or self._input_future.done()
395 ):
396 raise ActionUnavailable("The setup flow is not awaiting form input")
397 self.last_activity = time.monotonic()
398 errors: dict[str, str] = {}
399 parsed: dict[str, ConfigValueType] = {}
400 # gates resolve against the submitted values, so flipping one takes effect on the
401 # same submit regardless of where it sits on the form
402 submitted_entries = [
403 replace(entry, value=values.get(entry.key, entry.value)) for entry in step.entries
404 ]
405 for entry in step.entries:
406 if entry.type in UI_ONLY:
407 continue
408 raw_value = values.get(entry.key, entry.value)
409 try:
410 # parse_value also runs the entry's optional validate callback and
411 # stores the parsed value on the entry (echoed on a re-render)
412 # an entry behind an unmet dependency renders disabled, so demanding a
413 # value the user has no way to supply would wedge the flow
414 parsed[entry.key] = entry.parse_value(
415 raw_value, allow_none=not entry.dependency_met(submitted_entries)
416 )
417 except TypeError, ValueError:
418 errors[entry.key] = "required" if raw_value in (None, "") else "invalid_value"
419 if not isinstance(raw_value, list):
420 entry.value = raw_value
421 if entry.type == ConfigEntryType.SECURE_STRING:
422 # secure values are only handed to the flow coroutine; they are never
423 # kept on (or echoed back with) the stored step
424 entry.value = None
425 if errors:
426 step.errors = errors
427 self._publish_step(step)
428 return step
429 step.errors = {}
430 # clear the step-changed marker before waking the coroutine, so the submit
431 # caller can await the *next* published step without racing the coroutine
432 self._step_changed.clear()
433 self._input_future.set_result(parsed)
434 return None
435
436 async def handle_callback(self, request: Request) -> Response:
437 """
438 Handle an incoming request on this flow's external-step callback route.
439
440 :param request: The incoming (GET or POST) request; query parameters and any
441 POST body (JSON or form encoded) are merged into the callback params.
442 """
443 params: dict[str, str] = dict(request.query)
444 if request.method == "POST" and request.can_read_body:
445 try:
446 if request.content_type == "application/json":
447 body = json_loads(await request.read())
448 if isinstance(body, dict):
449 # coerce to str so the declared params contract holds for
450 # json bodies carrying numbers/bools
451 params.update({str(key): str(val) for key, val in body.items()})
452 else:
453 LOGGER.error("Ignoring non-object JSON setup flow callback body")
454 else:
455 params.update({key: str(val) for key, val in (await request.post()).items()})
456 except Exception as err:
457 LOGGER.error("Failed to parse setup flow callback body: %s", err)
458 if self._callback_future is not None and not self._callback_future.done():
459 # the values carry the authorization code/token, so only log the keys
460 LOGGER.debug(
461 "Setup flow %s resumed by callback with params: %s",
462 self.flow_id,
463 ", ".join(sorted(params)),
464 )
465 # only a callback that resolves a pending external step counts as
466 # activity: the route is unauthenticated, so bare requests must not
467 # be able to keep the flow alive past the idle TTL
468 self.last_activity = time.monotonic()
469 self._callback_future.set_result(params)
470 else:
471 LOGGER.debug(
472 "Received setup flow callback for %s while no external step is pending",
473 self.flow_id,
474 )
475 return Response(body=CALLBACK_RESPONSE_HTML, headers={"content-type": "text/html"})
476
477 async def wait_for_step_change(self, timeout: float) -> None:
478 """
479 Wait (bounded) until the flow publishes a step, returning silently on timeout.
480
481 :param timeout: Maximum time in seconds to wait.
482 """
483 try:
484 async with asyncio.timeout(timeout):
485 await self._step_changed.wait()
486 except TimeoutError:
487 return
488
489 def publish_abort(self, reason: str) -> None:
490 """
491 Publish the terminal ABORT step for this flow.
492
493 :param reason: Slug describing why the flow was aborted; resolved from the
494 translations (setup_flow.abort.<reason>) when the step is served.
495 """
496 step = self._build_step(FlowStepType.ABORT, "abort", reason=reason)
497 self._publish_step(step)
498
499 def close(self) -> None:
500 """Release the session's engine resources (callback route); called on flow end."""
501 if self._unregister_callback_route is not None:
502 self._unregister_callback_route()
503 self._unregister_callback_route = None
504
505 async def _await_with_deadline(self, awaitable: Awaitable[_T], expires_in: float | None) -> _T:
506 """
507 Await the given awaitable, converting the step deadline into StepExpiredError.
508
509 asyncio.timeout cancels the inner await at the deadline and raises TimeoutError
510 in *this* (the flow's) coroutine - exactly the required inject-into-the-author
511 semantic: catch it right at the awaiting call to refresh the step, or let it
512 propagate so the engine converts it into a timed_out ABORT.
513 """
514 if expires_in is None:
515 return await awaitable
516 try:
517 async with asyncio.timeout(expires_in):
518 return await awaitable
519 except TimeoutError as err:
520 raise StepExpiredError from err
521
522 def _build_step( # noqa: PLR0913 - mirrors the step model's fields
523 self,
524 step_type: FlowStepType,
525 step_id: str,
526 *,
527 entries: list[ConfigEntry] | None = None,
528 errors: dict[str, str] | None = None,
529 last_step: bool | None = None,
530 url: str | None = None,
531 progress_text: str | None = None,
532 progress: float | None = None,
533 image: str | None = None,
534 result: dict[str, str] | None = None,
535 reason: str | None = None,
536 translation_params: list[str] | None = None,
537 expires_in: float | None = None,
538 ) -> SetupFlowStep:
539 """Build a SetupFlowStep for this flow, stamping owner and absolute deadline."""
540 return SetupFlowStep(
541 flow_id=self.flow_id,
542 step_id=step_id,
543 type=step_type,
544 entries=entries or [],
545 errors=errors or {},
546 last_step=last_step,
547 url=url,
548 progress_text=progress_text,
549 progress=progress,
550 image=image,
551 # absolute epoch deadline so the remaining time survives a client re-fetch
552 expires_at=time.time() + expires_in if expires_in is not None else None,
553 result=result,
554 reason=reason,
555 translation_owner=self._translation_owner,
556 translation_params=translation_params,
557 )
558
559 def _prepare_entries(self, entries: list[ConfigEntry]) -> list[ConfigEntry]:
560 """Return copies of the form entries, validated and stamped for this flow."""
561 prepared: list[ConfigEntry] = []
562 for entry in entries:
563 if entry.action or entry.type == ConfigEntryType.ACTION:
564 # actions are the pseudo-flow mechanism of the options surface;
565 # inside real flows they are structurally banned
566 msg = f"Config entry {entry.key} is an action entry, not allowed in setup flows"
567 raise ValueError(msg)
568 # replace() returns a copy so the (often module-level) entry definitions
569 # are never mutated by submit-side value/owner stamping
570 copied = replace(
571 entry, translation_owner=entry.translation_owner or self._translation_owner
572 )
573 if copied.type == ConfigEntryType.SECURE_STRING:
574 # secrets never leave the server: a (prefilled) secure value is
575 # stripped from the step; the user always (re)types it
576 copied.value = None
577 prepared.append(copied)
578 return prepared
579
580 def _publish_step(self, step: SetupFlowStep) -> None:
581 """Store the step as current, mark activity and push it to subscribers."""
582 LOGGER.debug("Setup flow %s published %s step: %s", self.flow_id, step.type, step.step_id)
583 self.current_step = step
584 self.last_activity = time.monotonic()
585 self._step_changed.set()
586 self.mass.signal_event(EventType.SETUP_FLOW_UPDATED, object_id=self.flow_id, data=step)
587
588 def _ensure_callback_route(self) -> None:
589 """Register the flow's dynamic callback route (idempotent)."""
590 if self._unregister_callback_route is not None:
591 return
592 self._unregister_callback_route = self.mass.webserver.register_dynamic_route(
593 self._callback_path, self.handle_callback
594 )
595