From e82c14ec993361b5ee8511e30e44d71320d5c8c3 Mon Sep 17 00:00:00 2001 From: Andre Basche Date: Thu, 13 Apr 2023 23:25:49 +0200 Subject: [PATCH] Add some type hints --- pyhon/connection/api.py | 135 +++++++++++++++++++++++------------- pyhon/connection/auth.py | 3 +- pyhon/connection/device.py | 25 +++---- pyhon/connection/handler.py | 76 +++++++++++++------- pyhon/exceptions.py | 8 +++ pyhon/hon.py | 27 +++++--- 6 files changed, 177 insertions(+), 97 deletions(-) diff --git a/pyhon/connection/api.py b/pyhon/connection/api.py index dfa2b65..7704a22 100644 --- a/pyhon/connection/api.py +++ b/pyhon/connection/api.py @@ -1,46 +1,75 @@ import json import logging from datetime import datetime +from typing import Dict, Optional +from typing_extensions import Self -from pyhon import const +from aiohttp import ClientSession + +from pyhon import const, exceptions from pyhon.appliance import HonAppliance +from pyhon.connection.auth import HonAuth from pyhon.connection.handler import HonConnectionHandler, HonAnonymousConnectionHandler _LOGGER = logging.getLogger() class HonAPI: - def __init__(self, email="", password="", anonymous=False, session=None) -> None: + def __init__( + self, + email: str = "", + password: str = "", + anonymous: bool = False, + session: Optional[ClientSession] = None, + ) -> None: super().__init__() - self._email = email - self._password = password - self._anonymous = anonymous - self._hon = None - self._hon_anonymous = None - self._session = session + self._email: str = email + self._password: str = password + self._anonymous: bool = anonymous + self._hon_handler: Optional[HonConnectionHandler] = None + self._hon_anonymous_handler: Optional[HonAnonymousConnectionHandler] = None + self._session: Optional[ClientSession] = session - async def __aenter__(self): + async def __aenter__(self) -> Self: return await self.create() - async def __aexit__(self, exc_type, exc_val, exc_tb): + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: await self.close() - async def create(self): - self._hon_anonymous = await HonAnonymousConnectionHandler( + @property + def auth(self) -> HonAuth: + if self._hon is None or self._hon.auth is None: + raise exceptions.NoAuthenticationException + return self._hon.auth + + @property + def _hon(self): + if self._hon_handler is None: + raise exceptions.NoAuthenticationException + return self._hon_handler + + @property + def _hon_anonymous(self): + if self._hon_anonymous_handler is None: + raise exceptions.NoAuthenticationException + return self._hon_anonymous_handler + + async def create(self) -> Self: + self._hon_anonymous_handler = await HonAnonymousConnectionHandler( self._session ).create() if not self._anonymous: - self._hon = await HonConnectionHandler( + self._hon_handler = await HonConnectionHandler( self._email, self._password, self._session ).create() return self - async def load_appliances(self): + async def load_appliances(self) -> Dict: async with self._hon.get(f"{const.API_URL}/commands/v1/appliance") as resp: return await resp.json() - async def load_commands(self, appliance: HonAppliance): - params = { + async def load_commands(self, appliance: HonAppliance) -> Dict: + params: Dict = { "applianceType": appliance.appliance_type, "code": appliance.info["code"], "applianceModelId": appliance.appliance_model_id, @@ -51,52 +80,60 @@ class HonAPI: "appVersion": const.APP_VERSION, "series": appliance.info["series"], } - url = f"{const.API_URL}/commands/v1/retrieve" + url: str = f"{const.API_URL}/commands/v1/retrieve" async with self._hon.get(url, params=params) as response: - result = (await response.json()).get("payload", {}) + result: Dict = (await response.json()).get("payload", {}) if not result or result.pop("resultCode") != "0": return {} return result - async def command_history(self, appliance: HonAppliance): - url = f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/history" + async def command_history(self, appliance: HonAppliance) -> Dict: + url: str = ( + f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/history" + ) async with self._hon.get(url) as response: - result = await response.json() + result: Dict = await response.json() if not result or not result.get("payload"): return {} return result["payload"]["history"] - async def last_activity(self, appliance: HonAppliance): - url = f"{const.API_URL}/commands/v1/retrieve-last-activity" - params = {"macAddress": appliance.mac_address} + async def last_activity(self, appliance: HonAppliance) -> Dict: + url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity" + params: Dict = {"macAddress": appliance.mac_address} async with self._hon.get(url, params=params) as response: - result = await response.json() + result: Dict = await response.json() if result and (activity := result.get("attributes")): return activity return {} - async def load_attributes(self, appliance: HonAppliance): - params = { + async def load_attributes(self, appliance: HonAppliance) -> Dict: + params: Dict = { "macAddress": appliance.mac_address, "applianceType": appliance.appliance_type, "category": "CYCLE", } - url = f"{const.API_URL}/commands/v1/context" + url: str = f"{const.API_URL}/commands/v1/context" async with self._hon.get(url, params=params) as response: return (await response.json()).get("payload", {}) - async def load_statistics(self, appliance: HonAppliance): - params = { + async def load_statistics(self, appliance: HonAppliance) -> Dict: + params: Dict = { "macAddress": appliance.mac_address, "applianceType": appliance.appliance_type, } - url = f"{const.API_URL}/commands/v1/statistics" + url: str = f"{const.API_URL}/commands/v1/statistics" async with self._hon.get(url, params=params) as response: return (await response.json()).get("payload", {}) - async def send_command(self, appliance, command, parameters, ancillary_parameters): - now = datetime.utcnow().isoformat() - data = { + async def send_command( + self, + appliance: HonAppliance, + command: str, + parameters: Dict, + ancillary_parameters: Dict, + ) -> bool: + now: str = datetime.utcnow().isoformat() + data: Dict = { "macAddress": appliance.mac_address, "timestamp": f"{now[:-3]}Z", "commandName": command, @@ -112,36 +149,36 @@ class HonAPI: "parameters": parameters, "applianceType": appliance.appliance_type, } - url = f"{const.API_URL}/commands/v1/send" + url: str = f"{const.API_URL}/commands/v1/send" async with self._hon.post(url, json=data) as resp: - json_data = await resp.json() + json_data: Dict = await resp.json() if json_data.get("payload", {}).get("resultCode") == "0": return True return False - async def appliance_configuration(self): - url = f"{const.API_URL}/config/v1/appliance-configuration" + async def appliance_configuration(self) -> Dict: + url: str = f"{const.API_URL}/config/v1/appliance-configuration" async with self._hon_anonymous.get(url) as response: - result = await response.json() + result: Dict = await response.json() if result and (data := result.get("payload")): return data return {} - async def app_config(self, language="en", beta=True): - url = f"{const.API_URL}/app-config" - payload = { + async def app_config(self, language: str = "en", beta: bool = True) -> Dict: + url: str = f"{const.API_URL}/app-config" + payload_data: Dict = { "languageCode": language, "beta": beta, "appVersion": const.APP_VERSION, "os": const.OS, } - payload = json.dumps(payload, separators=(",", ":")) + payload: str = json.dumps(payload_data, separators=(",", ":")) async with self._hon_anonymous.post(url, data=payload) as response: if (result := await response.json()) and (data := result.get("payload")): return data return {} - async def translation_keys(self, language="en"): + async def translation_keys(self, language: str = "en") -> Dict: config = await self.app_config(language=language) if url := config.get("language", {}).get("jsonPath"): async with self._hon_anonymous.get(url) as response: @@ -149,8 +186,8 @@ class HonAPI: return result return {} - async def close(self): - if self._hon: - await self._hon.close() - if self._hon_anonymous: - await self._hon_anonymous.close() + async def close(self) -> None: + if self._hon_handler is not None: + await self._hon_handler.close() + if self._hon_anonymous_handler is not None: + await self._hon_anonymous_handler.close() diff --git a/pyhon/connection/auth.py b/pyhon/connection/auth.py index c488fc7..1cc0217 100644 --- a/pyhon/connection/auth.py +++ b/pyhon/connection/auth.py @@ -4,6 +4,7 @@ import re import secrets import urllib from pprint import pformat +from typing import List, Tuple from urllib import parse from urllib.parse import quote @@ -24,7 +25,7 @@ class HonAuth: self._cognito_token = "" self._id_token = "" self._device = device - self._called_urls = [] + self._called_urls: List[Tuple[int, str]] = [] @property def cognito_token(self): diff --git a/pyhon/connection/device.py b/pyhon/connection/device.py index f7399d9..39e529c 100644 --- a/pyhon/connection/device.py +++ b/pyhon/connection/device.py @@ -1,37 +1,38 @@ import secrets +from typing import Dict from pyhon import const class HonDevice: - def __init__(self): - self._app_version = const.APP_VERSION - self._os_version = const.OS_VERSION - self._os = const.OS - self._device_model = const.DEVICE_MODEL - self._mobile_id = secrets.token_hex(8) + def __init__(self) -> None: + self._app_version: str = const.APP_VERSION + self._os_version: int = const.OS_VERSION + self._os: str = const.OS + self._device_model: str = const.DEVICE_MODEL + self._mobile_id: str = secrets.token_hex(8) @property - def app_version(self): + def app_version(self) -> str: return self._app_version @property - def os_version(self): + def os_version(self) -> int: return self._os_version @property - def os(self): + def os(self) -> str: return self._os @property - def device_model(self): + def device_model(self) -> str: return self._device_model @property - def mobile_id(self): + def mobile_id(self) -> str: return self._mobile_id - def get(self): + def get(self) -> Dict: return { "appVersion": self.app_version, "mobileId": self.mobile_id, diff --git a/pyhon/connection/handler.py b/pyhon/connection/handler.py index 53e45a5..935c614 100644 --- a/pyhon/connection/handler.py +++ b/pyhon/connection/handler.py @@ -1,73 +1,93 @@ import json +from collections.abc import Generator, AsyncIterator, Coroutine from contextlib import asynccontextmanager +from typing import Optional, Callable, Dict +from typing_extensions import Self import aiohttp -from pyhon import const +from pyhon import const, exceptions from pyhon.connection.auth import HonAuth, _LOGGER from pyhon.connection.device import HonDevice from pyhon.exceptions import HonAuthenticationError class HonBaseConnectionHandler: - _HEADERS = {"user-agent": const.USER_AGENT, "Content-Type": "application/json"} + _HEADERS: Dict = { + "user-agent": const.USER_AGENT, + "Content-Type": "application/json", + } - def __init__(self, session=None): - self._create_session = session is None - self._session = session - self._auth = None + def __init__(self, session: Optional[aiohttp.ClientSession] = None) -> None: + self._create_session: bool = session is None + self._session: Optional[aiohttp.ClientSession] = session + self._auth: Optional[HonAuth] = None - async def __aenter__(self): + async def __aenter__(self) -> Self: return await self.create() - async def __aexit__(self, exc_type, exc_val, exc_tb): + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: await self.close() - async def create(self): + @property + def auth(self) -> Optional[HonAuth]: + return self._auth + + async def create(self) -> Self: if self._create_session: self._session = aiohttp.ClientSession() return self @asynccontextmanager - async def _intercept(self, method, *args, loop=0, **kwargs): + def _intercept(self, method: Callable, *args, loop: int = 0, **kwargs): raise NotImplementedError @asynccontextmanager - async def get(self, *args, **kwargs): + async def get(self, *args, **kwargs) -> AsyncIterator[Callable]: + if self._session is None: + raise exceptions.NoSessionException() + response: Callable async with self._intercept(self._session.get, *args, **kwargs) as response: yield response @asynccontextmanager - async def post(self, *args, **kwargs): + async def post(self, *args, **kwargs) -> AsyncIterator[Callable]: + if self._session is None: + raise exceptions.NoSessionException() + response: Callable async with self._intercept(self._session.post, *args, **kwargs) as response: yield response - async def close(self): - if self._create_session: + async def close(self) -> None: + if self._create_session and self._session is not None: await self._session.close() class HonConnectionHandler(HonBaseConnectionHandler): - def __init__(self, email, password, session=None): + def __init__( + self, email: str, password: str, session: Optional[aiohttp.ClientSession] = None + ) -> None: super().__init__(session=session) - self._device = HonDevice() - self._email = email - self._password = password + self._device: HonDevice = HonDevice() + self._email: str = email + self._password: str = password if not self._email: raise HonAuthenticationError("An email address must be specified") if not self._password: raise HonAuthenticationError("A password address must be specified") @property - def device(self): + def device(self) -> HonDevice: return self._device - async def create(self): + async def create(self) -> Self: await super().create() - self._auth = HonAuth(self._session, self._email, self._password, self._device) + self._auth: HonAuth = HonAuth( + self._session, self._email, self._password, self._device + ) return self - async def _check_headers(self, headers): + async def _check_headers(self, headers: Dict) -> Dict: if not (self._auth.cognito_token and self._auth.id_token): await self._auth.authenticate() headers["cognito-token"] = self._auth.cognito_token @@ -75,7 +95,9 @@ class HonConnectionHandler(HonBaseConnectionHandler): return self._HEADERS | headers @asynccontextmanager - async def _intercept(self, method, *args, loop=0, **kwargs): + async def _intercept( + self, method: Callable, *args, loop: int = 0, **kwargs + ) -> AsyncIterator: kwargs["headers"] = await self._check_headers(kwargs.get("headers", {})) async with method(*args, **kwargs) as response: if response.status in [401, 403] and loop == 0: @@ -116,14 +138,16 @@ class HonConnectionHandler(HonBaseConnectionHandler): response.status, await response.text(), ) - yield {} + raise HonAuthenticationError("Decode Error") class HonAnonymousConnectionHandler(HonBaseConnectionHandler): - _HEADERS = HonBaseConnectionHandler._HEADERS | {"x-api-key": const.API_KEY} + _HEADERS: Dict = HonBaseConnectionHandler._HEADERS | {"x-api-key": const.API_KEY} @asynccontextmanager - async def _intercept(self, method, *args, loop=0, **kwargs): + async def _intercept( + self, method: Callable, *args, loop: int = 0, **kwargs + ) -> AsyncIterator: kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS async with method(*args, **kwargs) as response: if response.status == 403: diff --git a/pyhon/exceptions.py b/pyhon/exceptions.py index 2e6a332..64e8648 100644 --- a/pyhon/exceptions.py +++ b/pyhon/exceptions.py @@ -4,3 +4,11 @@ class HonAuthenticationError(Exception): class HonNoAuthenticationNeeded(Exception): pass + + +class NoSessionException(Exception): + pass + + +class NoAuthenticationException(Exception): + pass diff --git a/pyhon/hon.py b/pyhon/hon.py index 322b077..64ac364 100644 --- a/pyhon/hon.py +++ b/pyhon/hon.py @@ -1,17 +1,20 @@ import asyncio -from typing import List +from typing import List, Optional +from typing_extensions import Self -from pyhon import HonAPI +from aiohttp import ClientSession + +from pyhon import HonAPI, exceptions from pyhon.appliance import HonAppliance class Hon: - def __init__(self, email, password, session=None): - self._email = email - self._password = password - self._session = session - self._appliances = [] - self._api = None + def __init__(self, email: str, password: str, session: ClientSession | None = None): + self._email: str = email + self._password: str = password + self._session: ClientSession | None = session + self._appliances: List[HonAppliance] = [] + self._api: Optional[HonAPI] = None async def __aenter__(self): return await self.create() @@ -19,7 +22,13 @@ class Hon: async def __aexit__(self, exc_type, exc_val, exc_tb): await self.close() - async def create(self): + @property + def api(self) -> HonAPI: + if self._api is None: + raise exceptions.NoAuthenticationException + return self._api + + async def create(self) -> Self: self._api = await HonAPI( self._email, self._password, session=self._session ).create()