Skip to content
Back to top

# Source code for aiohomeconnect.cli.client

    """Provide a CLI client for Home Connect API."""

    from __future__ import annotations

    import asyncio
    import contextlib
    import json
    from pathlib import Path
    import time
    from typing import Any

    from authlib.integrations.httpx_client import AsyncOAuth2Client
    from httpx import AsyncClient

    from aiohomeconnect.client import AbstractAuth, Client
    from aiohomeconnect.const import API_ENDPOINT, OAUTH2_AUTHORIZE, OAUTH2_TOKEN

    TOKEN_FILE = "token.json"  # noqa: S105
    TOKEN_EXPIRES_MARGIN = 20

    class CLIClient(Client):
        """Represent a CLI client for Home Connect API."""

        def __init__(
            self,
            client_id: str,
            client_secret: str,
            redirect_uri: str | None = None,
            scope: str | None = None,
        ) -> None:
            """Initialize the client."""
            super().__init__(
                Auth(
                    AsyncClient(),
                    API_ENDPOINT,
                    TokenManager(
                        client_id=client_id,
                        client_secret=client_secret,
                        redirect_uri=redirect_uri,
                        scope=scope,
                    ),
                ),
            )

    class Auth(AbstractAuth):
        """Implement the authentication."""

        def __init__(
            self,
            httpx_client: AsyncClient,
            host: str,
            token_manager: TokenManager,
        ) -> None:
            """Initialize the auth."""
            super().__init__(httpx_client, host)
            self.token_manager = token_manager

        async def async_get_access_token(self) -> str:
            """Return a valid access token."""
            if self.token_manager.access_token is None:
                await self.token_manager.load_access_token()
            if self.token_manager.access_token is None:
                raise ValueError("No access token available")
            if self.token_manager.is_token_valid():
                return self.token_manager.access_token

            await self.token_manager.refresh_access_token()
            await self.token_manager.save_access_token()

            return self.token_manager.access_token

    class TokenManager:
        """Manage the tokens for authentication."""

        def __init__(
            self,
            client_id: str,
            client_secret: str,
            redirect_uri: str | None = None,
            scope: str | None = None,
        ) -> None:
            """Initialize the token manager."""
            self.access_token: str | None = None
            self._refresh_token: str | None = None
            self._state: str | None = None
            self._token: dict[str, Any] = {}
            self._client = AsyncOAuth2Client(
                client_id,
                client_secret,
                scope=scope,
                redirect_uri=redirect_uri,
            )

        async def create_authorization_url(self) -> str:
            """Create the authorization URL."""
            uri, self._state = self._client.create_authorization_url(
                OAUTH2_AUTHORIZE,
            )
            return uri

        def is_token_valid(self) -> bool:
            """Check if the token is valid."""
            return self._token["expires_at"] > time.time() + TOKEN_EXPIRES_MARGIN

        async def fetch_access_token(self, code: str) -> dict[str, Any]:
            """Fetch the access token."""
            token = self._token = await self._client.fetch_token(
                OAUTH2_TOKEN,
                code=code,
                grant_type="authorization_code",
                state=self._state,
            )
            self._validate_token()
            self.access_token = token["access_token"]
            await self.save_access_token()
            return token

        async def load_access_token(self) -> None:
            """Load the access token."""
            await asyncio.to_thread(self._load_access_token)
            self.access_token = self._token.get("access_token")
            self._refresh_token = self._token.get("refresh_token")

        async def refresh_access_token(self) -> None:
            """Refresh the access token."""
            token = self._token = await self._client.refresh_token(
                OAUTH2_TOKEN,
                refresh_token=self._refresh_token,
                client_id=self._client.client_id,
                client_secret=self._client.client_secret,
            )
            self._validate_token()
            self.access_token = token["access_token"]
            self._refresh_token = token["refresh_token"]

        async def save_access_token(self) -> None:
            """Save the access token."""
            await asyncio.to_thread(self._save_access_token)

        def _load_access_token(self) -> None:
            """Load the access token."""
            with contextlib.suppress(FileNotFoundError):
                self._token = json.loads(Path(TOKEN_FILE).read_text(encoding="utf-8"))

        def _save_access_token(self) -> None:
            """Save the access token."""
            Path(TOKEN_FILE).write_text(json.dumps(self._token, indent=2), encoding="utf-8")

        def _validate_token(self) -> None:
            """Validate the token."""
            token = self._token
            token["expires_in"] = int(token["expires_in"])
            token["expires_at"] = time.time() + token["expires_in"]