/
/
/
1"""Setup flow for the Open Subsonic 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_PASSWORD, CONF_PATH, CONF_PORT, CONF_USERNAME
12from music_assistant.models.setup_flow import SetupFlowError
13from music_assistant.providers.opensubsonic.sonic_provider import CONF_API_KEY, CONF_BASE_URL
14
15if TYPE_CHECKING:
16 from music_assistant.models.setup_flow import SetupSession
17
18_ENTRIES = (
19 ConfigEntry(key=CONF_BASE_URL, type=ConfigEntryType.STRING, required=True),
20 ConfigEntry(key=CONF_API_KEY, type=ConfigEntryType.SECURE_STRING, required=False),
21 ConfigEntry(key=CONF_USERNAME, type=ConfigEntryType.STRING, required=False),
22 ConfigEntry(key=CONF_PASSWORD, type=ConfigEntryType.SECURE_STRING, required=False),
23 ConfigEntry(key=CONF_PORT, type=ConfigEntryType.INTEGER, required=False),
24 ConfigEntry(key=CONF_PATH, type=ConfigEntryType.STRING, required=False),
25)
26
27
28async def run_setup(session: SetupSession) -> None:
29 """Run the setup flow: collect the connection details and create the provider."""
30 errors: dict[str, str] | None = None
31 setup_data = dict(session.context.setup_data)
32 while True:
33 entries = [
34 replace(entry, value=setup_data.get(entry.key, entry.value)) for entry in _ENTRIES
35 ]
36 submitted = await session.form(entries, step_id="user", errors=errors, last_step=True)
37 setup_data.update(submitted)
38 try:
39 await session.finish(setup_data)
40 return
41 except SetupFlowError as err:
42 errors = {"base": err.translation_key or str(err)}
43