config_flow.py

Go to the documentation of this file.

"""Config flow for GitHub integration."""
 
from __future__ import annotations
 
import asyncio
from typing import TYPE_CHECKING, Any
 
from aiogithubapi import (
    GitHubAPI,
    GitHubDeviceAPI,
    GitHubException,
    GitHubLoginDeviceModel,
    GitHubLoginOauthModel,
)
from aiogithubapi.const import OAUTH_USER_LOGIN
import voluptuous as vol
 
from homeassistant.config_entries import (
    ConfigEntry,
    ConfigFlow,
    ConfigFlowResult,
    OptionsFlow,
)
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.aiohttp_client import (
    SERVER_SOFTWARE,
    async_get_clientsession,
)
import homeassistant.helpers.config_validation as cv
 
from .const import CLIENT_ID, CONF_REPOSITORIES, DEFAULT_REPOSITORIES, DOMAIN, LOGGER
 
 
async def get_repositories(hass: HomeAssistant, access_token: str) -> list[str]:
    """Return a list of repositories that the user owns or has starred."""
    client = GitHubAPI(token=access_token, session=async_get_clientsession(hass))
    repositories = set()
 
    async def _get_starred_repositories() -> None:
        response = await client.user.starred(params={"per_page": 100})
        if not response.is_last_page:
            results = await asyncio.gather(
                *(
                    client.user.starred(
                        params={"per_page": 100, "page": page_number},
                    )
                    for page_number in range(
                        response.next_page_number, response.last_page_number + 1
                    )
                )
            )
            for result in results:
                response.data.extend(result.data)
 
        repositories.update(response.data)
 
    async def _get_personal_repositories() -> None:
        response = await client.user.repos(params={"per_page": 100})
        if not response.is_last_page:
            results = await asyncio.gather(
                *(
                    client.user.repos(
                        params={"per_page": 100, "page": page_number},
                    )
                    for page_number in range(
                        response.next_page_number, response.last_page_number + 1
                    )
                )
            )
            for result in results:
                response.data.extend(result.data)
 
        repositories.update(response.data)
 
    try:
        await asyncio.gather(
            *(
                _get_starred_repositories(),
                _get_personal_repositories(),
            )
        )
 
    except GitHubException:
        return DEFAULT_REPOSITORIES
 
    if len(repositories) == 0:
        return DEFAULT_REPOSITORIES
 
    return sorted(
        (repo.full_name for repo in repositories),
        key=str.casefold,
    )
 
 
class GitHubConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for GitHub."""
 
    VERSION = 1
 
    login_task: asyncio.Task | None = None
 
    def __init__(self) -> None:
        """Initialize."""
        self._device_device: GitHubDeviceAPI | None = None
        self._login_login: GitHubLoginOauthModel | None = None
        self._login_device_login_device: GitHubLoginDeviceModel | None = None
 
    async def async_step_user(
        self,
        user_input: dict[str, Any] | None = None,
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        if self._async_current_entries_async_current_entries():
            return self.async_abortasync_abortasync_abort(reason="already_configured")
 
        return await self.async_step_deviceasync_step_device(user_input)
 
    async def async_step_device(
        self,
        user_input: dict[str, Any] | None = None,
    ) -> ConfigFlowResult:
        """Handle device steps."""
 
        async def _wait_for_login() -> None:
            if TYPE_CHECKING:
                # mypy is not aware that we can't get here without having these set already
                assert self._device_device is not None
                assert self._login_device_login_device is not None
 
            response = await self._device_device.activation(
                device_code=self._login_device_login_device.device_code
            )
            self._login_login = response.data
 
        if not self._device_device:
            self._device_device = GitHubDeviceAPI(
                client_id=CLIENT_ID,
                session=async_get_clientsession(self.hass),
                client_name=SERVER_SOFTWARE,
            )
 
            try:
                response = await self._device_device.register()
                self._login_device_login_device = response.data
            except GitHubException as exception:
                LOGGER.exception(exception)
                return self.async_abortasync_abortasync_abort(reason="could_not_register")
 
        if self.login_tasklogin_task is None:
            self.login_tasklogin_task = self.hass.async_create_task(_wait_for_login())
 
        if self.login_tasklogin_task.done():
            if self.login_tasklogin_task.exception():
                return self.async_show_progress_doneasync_show_progress_done(next_step_id="could_not_register")
            return self.async_show_progress_doneasync_show_progress_done(next_step_id="repositories")
 
        if TYPE_CHECKING:
            # mypy is not aware that we can't get here without having this set already
            assert self._login_device_login_device is not None
 
        return self.async_show_progressasync_show_progress(
            step_id="device",
            progress_action="wait_for_device",
            description_placeholders={
                "url": OAUTH_USER_LOGIN,
                "code": self._login_device_login_device.user_code,
            },
            progress_task=self.login_tasklogin_task,
        )
 
    async def async_step_repositories(
        self,
        user_input: dict[str, Any] | None = None,
    ) -> ConfigFlowResult:
        """Handle repositories step."""
 
        if TYPE_CHECKING:
            # mypy is not aware that we can't get here without having this set already
            assert self._login_login is not None
 
        if not user_input:
            repositories = await get_repositories(self.hass, self._login_login.access_token)
            return self.async_show_formasync_show_formasync_show_form(
                step_id="repositories",
                data_schema=vol.Schema(
                    {
                        vol.Required(CONF_REPOSITORIES): cv.multi_select(
                            {k: k for k in repositories}
                        ),
                    }
                ),
            )
 
        return self.async_create_entryasync_create_entryasync_create_entry(
            title="",
            data={CONF_ACCESS_TOKEN: self._login_login.access_token},
            options={CONF_REPOSITORIES: user_input[CONF_REPOSITORIES]},
        )
 
    async def async_step_could_not_register(
        self,
        user_input: dict[str, Any] | None = None,
    ) -> ConfigFlowResult:
        """Handle issues that need transition await from progress step."""
        return self.async_abortasync_abortasync_abort(reason="could_not_register")
 
    @staticmethod
    @callback
    def async_get_options_flow(
        config_entry: ConfigEntry,
    ) -> OptionsFlowHandler:
        """Get the options flow for this handler."""
        return OptionsFlowHandler()
 
 
class OptionsFlowHandler(OptionsFlow):
    """Handle a option flow for GitHub."""
 
    async def async_step_init(
        self,
        user_input: dict[str, Any] | None = None,
    ) -> ConfigFlowResult:
        """Handle options flow."""
        if not user_input:
            configured_repositories: list[str] = self.config_entryconfig_entryconfig_entry.options[
                CONF_REPOSITORIES
            ]
            repositories = await get_repositories(
                self.hass, self.config_entryconfig_entryconfig_entry.data[CONF_ACCESS_TOKEN]
            )
 
            # In case the user has removed a starred repository that is already tracked
            for repository in configured_repositories:
                if repository not in repositories:
                    repositories.append(repository)
 
            return self.async_show_formasync_show_form(
                step_id="init",
                data_schema=vol.Schema(
                    {
                        vol.Required(
                            CONF_REPOSITORIES,
                            default=configured_repositories,
                        ): cv.multi_select({k: k for k in repositories}),
                    }
                ),
            )
 
        return self.async_create_entryasync_create_entry(title="", data=user_input)