Improve august typing (2) (#108327)

pull/108329/head
Marc Mueller 2024-01-18 23:24:41 +01:00 committed by GitHub
parent 5f08e2a2d1
commit 72667adeba
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 22 additions and 10 deletions

View File

@ -2,11 +2,11 @@
from __future__ import annotations
import asyncio
from collections.abc import Iterable, ValuesView
from collections.abc import Callable, Coroutine, Iterable, ValuesView
from datetime import datetime
from itertools import chain
import logging
from typing import Any
from typing import Any, ParamSpec, TypeVar
from aiohttp import ClientError, ClientResponseError
from yalexs.const import DEFAULT_BRAND
@ -34,6 +34,9 @@ from .gateway import AugustGateway
from .subscriber import AugustSubscriberMixin
from .util import async_create_august_clientsession
_R = TypeVar("_R")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
API_CACHED_ATTRS = {
@ -360,7 +363,7 @@ class AugustData(AugustSubscriberMixin):
return device.device_name
return None
async def async_lock(self, device_id):
async def async_lock(self, device_id: str):
"""Lock the device."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -369,7 +372,9 @@ class AugustData(AugustSubscriberMixin):
device_id,
)
async def async_status_async(self, device_id, hyper_bridge):
async def async_status_async(
self, device_id: str, hyper_bridge: bool
) -> str | None:
"""Request status of the device but do not wait for a response since it will come via pubnub."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -379,7 +384,7 @@ class AugustData(AugustSubscriberMixin):
hyper_bridge,
)
async def async_lock_async(self, device_id, hyper_bridge):
async def async_lock_async(self, device_id: str, hyper_bridge: bool) -> str | None:
"""Lock the device but do not wait for a response since it will come via pubnub."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -389,7 +394,7 @@ class AugustData(AugustSubscriberMixin):
hyper_bridge,
)
async def async_unlock(self, device_id):
async def async_unlock(self, device_id: str):
"""Unlock the device."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -398,7 +403,9 @@ class AugustData(AugustSubscriberMixin):
device_id,
)
async def async_unlock_async(self, device_id, hyper_bridge):
async def async_unlock_async(
self, device_id: str, hyper_bridge: bool
) -> str | None:
"""Unlock the device but do not wait for a response since it will come via pubnub."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -409,8 +416,12 @@ class AugustData(AugustSubscriberMixin):
)
async def _async_call_api_op_requires_bridge(
self, device_id, func, *args, **kwargs
):
self,
device_id: str,
func: Callable[_P, Coroutine[Any, Any, _R]],
*args: _P.args,
**kwargs: _P.kwargs,
) -> _R | None:
"""Call an API that requires the bridge to be online and will change the device state."""
ret = None
try:

View File

@ -34,6 +34,8 @@ _LOGGER = logging.getLogger(__name__)
class AugustGateway:
"""Handle the connection to August."""
api: ApiAsync
def __init__(self, hass: HomeAssistant, aiohttp_session: ClientSession) -> None:
"""Init the connection."""
self._aiohttp_session = aiohttp_session
@ -41,7 +43,6 @@ class AugustGateway:
self._access_token_cache_file: str | None = None
self._hass: HomeAssistant = hass
self._config: Mapping[str, Any] | None = None
self.api: ApiAsync | None = None
self.authenticator: AuthenticatorAsync | None = None
self.authentication: Authentication | None = None