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