Improve type hints in plex config flow (#124914)

This commit is contained in:
epenet 2024-08-30 11:22:07 +02:00 committed by GitHub
parent df2ea1e875
commit 19cbc1b258
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -35,7 +35,7 @@ from homeassistant.const import (
CONF_URL,
CONF_VERIFY_SSL,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
@ -71,7 +71,7 @@ _LOGGER = logging.getLogger(__package__)
@callback
def configured_servers(hass):
def configured_servers(hass: HomeAssistant) -> set[str]:
"""Return a set of the configured Plex servers."""
return {
entry.data[CONF_SERVER_IDENTIFIER]
@ -79,7 +79,7 @@ def configured_servers(hass):
}
async def async_discover(hass):
async def async_discover(hass: HomeAssistant) -> None:
"""Scan for available Plex servers."""
gdm = GDM()
await hass.async_add_executor_job(gdm.scan)
@ -97,6 +97,9 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1
available_servers: list[tuple[str, str, str]]
plexauth: PlexAuth
@staticmethod
@callback
def async_get_options_flow(
@ -108,28 +111,34 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize the Plex flow."""
self.current_login: dict[str, Any] = {}
self.available_servers = None
self.plexauth = None
self.token = None
self.client_id = None
self._manual = False
self._reauth_config: dict[str, Any] | None = None
async def async_step_user(self, user_input=None, errors=None):
async def async_step_user(
self,
user_input: dict[str, Any] | None = None,
errors: dict[str, str] | None = None,
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
if user_input is not None:
return await self.async_step_plex_website_auth()
return await self._async_step_plex_website_auth()
if self.show_advanced_options:
return await self.async_step_user_advanced(errors=errors)
return self.async_show_form(step_id="user", errors=errors)
async def async_step_user_advanced(self, user_input=None, errors=None):
async def async_step_user_advanced(
self,
user_input: dict[str, str] | None = None,
errors: dict[str, str] | None = None,
) -> ConfigFlowResult:
"""Handle an advanced mode flow initialized by the user."""
if user_input is not None:
if user_input.get("setup_method") == MANUAL_SETUP_STRING:
self._manual = True
return await self.async_step_manual_setup()
return await self.async_step_plex_website_auth()
return await self._async_step_plex_website_auth()
data_schema = vol.Schema(
{
@ -142,7 +151,11 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN):
step_id="user_advanced", data_schema=data_schema, errors=errors
)
async def async_step_manual_setup(self, user_input=None, errors=None):
async def async_step_manual_setup(
self,
user_input: dict[str, Any] | None = None,
errors: dict[str, str] | None = None,
) -> ConfigFlowResult:
"""Begin manual configuration."""
if user_input is not None and errors is None:
user_input.pop(CONF_URL, None)
@ -264,7 +277,9 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=url, data=data)
async def async_step_select_server(self, user_input=None):
async def async_step_select_server(
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Use selected Plex server."""
config = dict(self.current_login)
if user_input is not None:
@ -292,7 +307,9 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN):
errors={},
)
async def async_step_integration_discovery(self, discovery_info):
async def async_step_integration_discovery(
self, discovery_info: dict[str, Any]
) -> ConfigFlowResult:
"""Handle GDM discovery."""
machine_identifier = discovery_info["data"]["Resource-Identifier"]
await self.async_set_unique_id(machine_identifier)
@ -305,7 +322,7 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN):
}
return await self.async_step_user()
async def async_step_plex_website_auth(self):
async def _async_step_plex_website_auth(self) -> ConfigFlowResult:
"""Begin external auth flow on Plex website."""
self.hass.http.register_view(PlexAuthorizationCallbackView)
if (req := http.current_request.get()) is None:
@ -329,7 +346,9 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN):
auth_url = self.plexauth.auth_url(forward_url)
return self.async_external_step(step_id="obtain_token", url=auth_url)
async def async_step_obtain_token(self, user_input=None):
async def async_step_obtain_token(
self, user_input: None = None
) -> ConfigFlowResult:
"""Obtain token after external auth completed."""
token = await self.plexauth.token(10)
@ -340,11 +359,13 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN):
self.client_id = self.plexauth.client_identifier
return self.async_external_step_done(next_step_id="use_external_token")
async def async_step_timed_out(self, user_input=None):
async def async_step_timed_out(self, user_input: None = None) -> ConfigFlowResult:
"""Abort flow when time expires."""
return self.async_abort(reason="token_request_timeout")
async def async_step_use_external_token(self, user_input=None):
async def async_step_use_external_token(
self, user_input: None = None
) -> ConfigFlowResult:
"""Continue server validation with external token."""
server_config = {CONF_TOKEN: self.token}
return await self.async_step_server_validate(server_config)
@ -367,11 +388,13 @@ class PlexOptionsFlowHandler(OptionsFlow):
self.options = copy.deepcopy(dict(config_entry.options))
self.server_id = config_entry.data[CONF_SERVER_IDENTIFIER]
async def async_step_init(self, user_input=None):
async def async_step_init(self, user_input: None = None) -> ConfigFlowResult:
"""Manage the Plex options."""
return await self.async_step_plex_mp_settings()
async def async_step_plex_mp_settings(self, user_input=None):
async def async_step_plex_mp_settings(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the Plex media_player options."""
plex_server = get_plex_server(self.hass, self.server_id)