music-assistant-server

1"""Setup flow for the Youtube Music provider."""
2
3from __future__ import annotations
4
5from dataclasses import replace
6from typing import TYPE_CHECKING
7
8from music_assistant_models.config_entries import ConfigEntry
9from music_assistant_models.enums import ConfigEntryType
10
11from music_assistant.constants import CONF_USERNAME
12from music_assistant.models.setup_flow import SetupFlowError
13from music_assistant.providers.ytmusic import (
14    CONF_COOKIE,
15    CONF_PO_TOKEN_SERVER_URL,
16    DEFAULT_PO_TOKEN_SERVER_URL,
17)
18
19if TYPE_CHECKING:
20    from music_assistant.models.setup_flow import SetupSession
21
22_ENTRIES = (
23    ConfigEntry(key=CONF_USERNAME, type=ConfigEntryType.STRING, required=True),
24    ConfigEntry(
25        key=CONF_COOKIE,
26        type=ConfigEntryType.SECURE_STRING,
27        required=True,
28    ),
29    ConfigEntry(
30        key=CONF_PO_TOKEN_SERVER_URL,
31        type=ConfigEntryType.STRING,
32        default_value=DEFAULT_PO_TOKEN_SERVER_URL,
33        required=True,
34    ),
35)
36
37
38async def run_setup(session: SetupSession) -> None:
39    """Run the setup flow: collect the credentials and create the provider."""
40    errors: dict[str, str] | None = None
41    setup_data = dict(session.context.setup_data)
42    while True:
43        entries = [
44            replace(entry, value=setup_data.get(entry.key, entry.value)) for entry in _ENTRIES
45        ]
46        submitted = await session.form(entries, step_id="user", errors=errors, last_step=True)
47        setup_data.update(submitted)
48        try:
49            await session.finish(setup_data)
50            return
51        except SetupFlowError as err:
52            errors = {"base": err.translation_key or str(err)}
53