Add some type hints

This commit is contained in:
Andre Basche 2023-04-13 23:25:49 +02:00
parent 970b94bfa7
commit e82c14ec99
6 changed files with 177 additions and 97 deletions

View File

@ -1,46 +1,75 @@
import json import json
import logging import logging
from datetime import datetime 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.appliance import HonAppliance
from pyhon.connection.auth import HonAuth
from pyhon.connection.handler import HonConnectionHandler, HonAnonymousConnectionHandler from pyhon.connection.handler import HonConnectionHandler, HonAnonymousConnectionHandler
_LOGGER = logging.getLogger() _LOGGER = logging.getLogger()
class HonAPI: 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__() super().__init__()
self._email = email self._email: str = email
self._password = password self._password: str = password
self._anonymous = anonymous self._anonymous: bool = anonymous
self._hon = None self._hon_handler: Optional[HonConnectionHandler] = None
self._hon_anonymous = None self._hon_anonymous_handler: Optional[HonAnonymousConnectionHandler] = None
self._session = session self._session: Optional[ClientSession] = session
async def __aenter__(self): async def __aenter__(self) -> Self:
return await self.create() 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() await self.close()
async def create(self): @property
self._hon_anonymous = await HonAnonymousConnectionHandler( 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 self._session
).create() ).create()
if not self._anonymous: if not self._anonymous:
self._hon = await HonConnectionHandler( self._hon_handler = await HonConnectionHandler(
self._email, self._password, self._session self._email, self._password, self._session
).create() ).create()
return self 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: async with self._hon.get(f"{const.API_URL}/commands/v1/appliance") as resp:
return await resp.json() return await resp.json()
async def load_commands(self, appliance: HonAppliance): async def load_commands(self, appliance: HonAppliance) -> Dict:
params = { params: Dict = {
"applianceType": appliance.appliance_type, "applianceType": appliance.appliance_type,
"code": appliance.info["code"], "code": appliance.info["code"],
"applianceModelId": appliance.appliance_model_id, "applianceModelId": appliance.appliance_model_id,
@ -51,52 +80,60 @@ class HonAPI:
"appVersion": const.APP_VERSION, "appVersion": const.APP_VERSION,
"series": appliance.info["series"], "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: 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": if not result or result.pop("resultCode") != "0":
return {} return {}
return result return result
async def command_history(self, appliance: HonAppliance): async def command_history(self, appliance: HonAppliance) -> Dict:
url = f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/history" url: str = (
f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/history"
)
async with self._hon.get(url) as response: 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"): if not result or not result.get("payload"):
return {} return {}
return result["payload"]["history"] return result["payload"]["history"]
async def last_activity(self, appliance: HonAppliance): async def last_activity(self, appliance: HonAppliance) -> Dict:
url = f"{const.API_URL}/commands/v1/retrieve-last-activity" url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity"
params = {"macAddress": appliance.mac_address} params: Dict = {"macAddress": appliance.mac_address}
async with self._hon.get(url, params=params) as response: 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")): if result and (activity := result.get("attributes")):
return activity return activity
return {} return {}
async def load_attributes(self, appliance: HonAppliance): async def load_attributes(self, appliance: HonAppliance) -> Dict:
params = { params: Dict = {
"macAddress": appliance.mac_address, "macAddress": appliance.mac_address,
"applianceType": appliance.appliance_type, "applianceType": appliance.appliance_type,
"category": "CYCLE", "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: async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {}) return (await response.json()).get("payload", {})
async def load_statistics(self, appliance: HonAppliance): async def load_statistics(self, appliance: HonAppliance) -> Dict:
params = { params: Dict = {
"macAddress": appliance.mac_address, "macAddress": appliance.mac_address,
"applianceType": appliance.appliance_type, "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: async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {}) return (await response.json()).get("payload", {})
async def send_command(self, appliance, command, parameters, ancillary_parameters): async def send_command(
now = datetime.utcnow().isoformat() self,
data = { appliance: HonAppliance,
command: str,
parameters: Dict,
ancillary_parameters: Dict,
) -> bool:
now: str = datetime.utcnow().isoformat()
data: Dict = {
"macAddress": appliance.mac_address, "macAddress": appliance.mac_address,
"timestamp": f"{now[:-3]}Z", "timestamp": f"{now[:-3]}Z",
"commandName": command, "commandName": command,
@ -112,36 +149,36 @@ class HonAPI:
"parameters": parameters, "parameters": parameters,
"applianceType": appliance.appliance_type, "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: 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": if json_data.get("payload", {}).get("resultCode") == "0":
return True return True
return False return False
async def appliance_configuration(self): async def appliance_configuration(self) -> Dict:
url = f"{const.API_URL}/config/v1/appliance-configuration" url: str = f"{const.API_URL}/config/v1/appliance-configuration"
async with self._hon_anonymous.get(url) as response: 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")): if result and (data := result.get("payload")):
return data return data
return {} return {}
async def app_config(self, language="en", beta=True): async def app_config(self, language: str = "en", beta: bool = True) -> Dict:
url = f"{const.API_URL}/app-config" url: str = f"{const.API_URL}/app-config"
payload = { payload_data: Dict = {
"languageCode": language, "languageCode": language,
"beta": beta, "beta": beta,
"appVersion": const.APP_VERSION, "appVersion": const.APP_VERSION,
"os": const.OS, "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: async with self._hon_anonymous.post(url, data=payload) as response:
if (result := await response.json()) and (data := result.get("payload")): if (result := await response.json()) and (data := result.get("payload")):
return data return data
return {} return {}
async def translation_keys(self, language="en"): async def translation_keys(self, language: str = "en") -> Dict:
config = await self.app_config(language=language) config = await self.app_config(language=language)
if url := config.get("language", {}).get("jsonPath"): if url := config.get("language", {}).get("jsonPath"):
async with self._hon_anonymous.get(url) as response: async with self._hon_anonymous.get(url) as response:
@ -149,8 +186,8 @@ class HonAPI:
return result return result
return {} return {}
async def close(self): async def close(self) -> None:
if self._hon: if self._hon_handler is not None:
await self._hon.close() await self._hon_handler.close()
if self._hon_anonymous: if self._hon_anonymous_handler is not None:
await self._hon_anonymous.close() await self._hon_anonymous_handler.close()

View File

@ -4,6 +4,7 @@ import re
import secrets import secrets
import urllib import urllib
from pprint import pformat from pprint import pformat
from typing import List, Tuple
from urllib import parse from urllib import parse
from urllib.parse import quote from urllib.parse import quote
@ -24,7 +25,7 @@ class HonAuth:
self._cognito_token = "" self._cognito_token = ""
self._id_token = "" self._id_token = ""
self._device = device self._device = device
self._called_urls = [] self._called_urls: List[Tuple[int, str]] = []
@property @property
def cognito_token(self): def cognito_token(self):

View File

@ -1,37 +1,38 @@
import secrets import secrets
from typing import Dict
from pyhon import const from pyhon import const
class HonDevice: class HonDevice:
def __init__(self): def __init__(self) -> None:
self._app_version = const.APP_VERSION self._app_version: str = const.APP_VERSION
self._os_version = const.OS_VERSION self._os_version: int = const.OS_VERSION
self._os = const.OS self._os: str = const.OS
self._device_model = const.DEVICE_MODEL self._device_model: str = const.DEVICE_MODEL
self._mobile_id = secrets.token_hex(8) self._mobile_id: str = secrets.token_hex(8)
@property @property
def app_version(self): def app_version(self) -> str:
return self._app_version return self._app_version
@property @property
def os_version(self): def os_version(self) -> int:
return self._os_version return self._os_version
@property @property
def os(self): def os(self) -> str:
return self._os return self._os
@property @property
def device_model(self): def device_model(self) -> str:
return self._device_model return self._device_model
@property @property
def mobile_id(self): def mobile_id(self) -> str:
return self._mobile_id return self._mobile_id
def get(self): def get(self) -> Dict:
return { return {
"appVersion": self.app_version, "appVersion": self.app_version,
"mobileId": self.mobile_id, "mobileId": self.mobile_id,

View File

@ -1,73 +1,93 @@
import json import json
from collections.abc import Generator, AsyncIterator, Coroutine
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Optional, Callable, Dict
from typing_extensions import Self
import aiohttp import aiohttp
from pyhon import const from pyhon import const, exceptions
from pyhon.connection.auth import HonAuth, _LOGGER from pyhon.connection.auth import HonAuth, _LOGGER
from pyhon.connection.device import HonDevice from pyhon.connection.device import HonDevice
from pyhon.exceptions import HonAuthenticationError from pyhon.exceptions import HonAuthenticationError
class HonBaseConnectionHandler: 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): def __init__(self, session: Optional[aiohttp.ClientSession] = None) -> None:
self._create_session = session is None self._create_session: bool = session is None
self._session = session self._session: Optional[aiohttp.ClientSession] = session
self._auth = None self._auth: Optional[HonAuth] = None
async def __aenter__(self): async def __aenter__(self) -> Self:
return await self.create() 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() 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: if self._create_session:
self._session = aiohttp.ClientSession() self._session = aiohttp.ClientSession()
return self return self
@asynccontextmanager @asynccontextmanager
async def _intercept(self, method, *args, loop=0, **kwargs): def _intercept(self, method: Callable, *args, loop: int = 0, **kwargs):
raise NotImplementedError raise NotImplementedError
@asynccontextmanager @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: async with self._intercept(self._session.get, *args, **kwargs) as response:
yield response yield response
@asynccontextmanager @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: async with self._intercept(self._session.post, *args, **kwargs) as response:
yield response yield response
async def close(self): async def close(self) -> None:
if self._create_session: if self._create_session and self._session is not None:
await self._session.close() await self._session.close()
class HonConnectionHandler(HonBaseConnectionHandler): 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) super().__init__(session=session)
self._device = HonDevice() self._device: HonDevice = HonDevice()
self._email = email self._email: str = email
self._password = password self._password: str = password
if not self._email: if not self._email:
raise HonAuthenticationError("An email address must be specified") raise HonAuthenticationError("An email address must be specified")
if not self._password: if not self._password:
raise HonAuthenticationError("A password address must be specified") raise HonAuthenticationError("A password address must be specified")
@property @property
def device(self): def device(self) -> HonDevice:
return self._device return self._device
async def create(self): async def create(self) -> Self:
await super().create() 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 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): if not (self._auth.cognito_token and self._auth.id_token):
await self._auth.authenticate() await self._auth.authenticate()
headers["cognito-token"] = self._auth.cognito_token headers["cognito-token"] = self._auth.cognito_token
@ -75,7 +95,9 @@ class HonConnectionHandler(HonBaseConnectionHandler):
return self._HEADERS | headers return self._HEADERS | headers
@asynccontextmanager @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", {})) kwargs["headers"] = await self._check_headers(kwargs.get("headers", {}))
async with method(*args, **kwargs) as response: async with method(*args, **kwargs) as response:
if response.status in [401, 403] and loop == 0: if response.status in [401, 403] and loop == 0:
@ -116,14 +138,16 @@ class HonConnectionHandler(HonBaseConnectionHandler):
response.status, response.status,
await response.text(), await response.text(),
) )
yield {} raise HonAuthenticationError("Decode Error")
class HonAnonymousConnectionHandler(HonBaseConnectionHandler): class HonAnonymousConnectionHandler(HonBaseConnectionHandler):
_HEADERS = HonBaseConnectionHandler._HEADERS | {"x-api-key": const.API_KEY} _HEADERS: Dict = HonBaseConnectionHandler._HEADERS | {"x-api-key": const.API_KEY}
@asynccontextmanager @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 kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS
async with method(*args, **kwargs) as response: async with method(*args, **kwargs) as response:
if response.status == 403: if response.status == 403:

View File

@ -4,3 +4,11 @@ class HonAuthenticationError(Exception):
class HonNoAuthenticationNeeded(Exception): class HonNoAuthenticationNeeded(Exception):
pass pass
class NoSessionException(Exception):
pass
class NoAuthenticationException(Exception):
pass

View File

@ -1,17 +1,20 @@
import asyncio 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 from pyhon.appliance import HonAppliance
class Hon: class Hon:
def __init__(self, email, password, session=None): def __init__(self, email: str, password: str, session: ClientSession | None = None):
self._email = email self._email: str = email
self._password = password self._password: str = password
self._session = session self._session: ClientSession | None = session
self._appliances = [] self._appliances: List[HonAppliance] = []
self._api = None self._api: Optional[HonAPI] = None
async def __aenter__(self): async def __aenter__(self):
return await self.create() return await self.create()
@ -19,7 +22,13 @@ class Hon:
async def __aexit__(self, exc_type, exc_val, exc_tb): async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.close() 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._api = await HonAPI(
self._email, self._password, session=self._session self._email, self._password, session=self._session
).create() ).create()