diff --git a/pyoverkiz/client.py b/pyoverkiz/client.py index 510d6237..b34eb5a2 100644 --- a/pyoverkiz/client.py +++ b/pyoverkiz/client.py @@ -48,6 +48,7 @@ TooManyConcurrentRequestsException, TooManyExecutionsException, TooManyRequestsException, + UnknownException, ) from pyoverkiz.models import ( Command, @@ -119,7 +120,7 @@ def __init__( self.gateways: list[Gateway] = [] self.event_listener_id: str | None = None - self.session = session if session else ClientSession() + self.session = session or ClientSession() async def __aenter__(self) -> OverkizClient: return self @@ -186,7 +187,7 @@ async def somfy_tahoma_get_access_token(self) -> str: """ # Request access token async with self.session.post( - SOMFY_API + "/oauth/oauth/v2/token", + f"{SOMFY_API}/oauth/oauth/v2/token", data=FormData( { "grant_type": "password", @@ -230,7 +231,7 @@ async def refresh_token(self) -> None: # &grant_type=refresh_token&refresh_token=REFRESH_TOKEN # Request access token async with self.session.post( - SOMFY_API + "/oauth/oauth/v2/token", + f"{SOMFY_API}/oauth/oauth/v2/token", data=FormData( { "grant_type": "refresh_token", @@ -263,7 +264,7 @@ async def cozytouch_login(self) -> str: """ # Request access token async with self.session.post( - COZYTOUCH_ATLANTIC_API + "/token", + f"{COZYTOUCH_ATLANTIC_API}/token", data=FormData( { "grant_type": "password", @@ -288,7 +289,7 @@ async def cozytouch_login(self) -> str: # Request JWT async with self.session.get( - COZYTOUCH_ATLANTIC_API + "/gacoma/gacomawcfservice/accounts/jwt", + f"{COZYTOUCH_ATLANTIC_API}/gacoma/gacomawcfservice/accounts/jwt", headers={"Authorization": f"Bearer {token['access_token']}"}, ) as response: jwt = await response.text() @@ -298,7 +299,7 @@ async def cozytouch_login(self) -> str: jwt = jwt.strip('"') # Remove surrounding quotes - return jwt + return str(jwt) async def nexity_login(self) -> str: """ @@ -458,9 +459,7 @@ async def get_execution_history(self) -> list[HistoryExecution]: List execution history """ response = await self.__get("history/executions") - execution_history = [HistoryExecution(**h) for h in humps.decamelize(response)] - - return execution_history + return [HistoryExecution(**h) for h in humps.decamelize(response)] @backoff.on_exception( backoff.expo, NotAuthenticatedException, max_tries=2, on_backoff=relogin @@ -485,9 +484,7 @@ async def get_state(self, deviceurl: str) -> list[State]: response = await self.__get( f"setup/devices/{urllib.parse.quote_plus(deviceurl)}/states" ) - state = [State(**s) for s in humps.decamelize(response)] - - return state + return [State(**s) for s in humps.decamelize(response)] @backoff.on_exception( backoff.expo, NotAuthenticatedException, max_tries=2, on_backoff=relogin @@ -534,9 +531,7 @@ async def fetch_events(self) -> list[Event]: """ await self._refresh_token_if_expired() response = await self.__post(f"events/{self.event_listener_id}/fetch") - events = [Event(**e) for e in humps.decamelize(response)] - - return events + return [Event(**e) for e in humps.decamelize(response)] async def unregister_event_listener(self) -> None: """ @@ -553,9 +548,7 @@ async def unregister_event_listener(self) -> None: async def get_current_execution(self, exec_id: str) -> Execution: """Get an action group execution currently running""" response = await self.__get(f"exec/current/{exec_id}") - execution = Execution(**humps.decamelize(response)) - - return execution + return Execution(**humps.decamelize(response)) @backoff.on_exception( backoff.expo, NotAuthenticatedException, max_tries=2, on_backoff=relogin @@ -563,9 +556,7 @@ async def get_current_execution(self, exec_id: str) -> Execution: async def get_current_executions(self) -> list[Execution]: """Get all action groups executions currently running""" response = await self.__get("exec/current") - executions = [Execution(**e) for e in humps.decamelize(response)] - - return executions + return [Execution(**e) for e in humps.decamelize(response)] @backoff.on_exception(backoff.expo, TooManyExecutionsException, max_tries=10) @backoff.on_exception( @@ -629,8 +620,7 @@ async def get_scenarios(self) -> list[Scenario]: async def get_places(self) -> Place: """List the places""" response = await self.__get("setup/places") - places = Place(**humps.decamelize(response)) - return places + return Place(**humps.decamelize(response)) @backoff.on_exception( backoff.expo, @@ -681,9 +671,7 @@ async def get_local_tokens( Access scope : Full enduser API access (enduser/*) """ response = await self.__get(f"/config/{gateway_id}/local/tokens/{scope}") - local_tokens = [LocalToken(**lt) for lt in humps.decamelize(response)] - - return local_tokens + return [LocalToken(**lt) for lt in humps.decamelize(response)] @backoff.on_exception( backoff.expo, @@ -764,7 +752,7 @@ async def check_response(response: ClientResponse) -> None: result = await response.text() if "Server is down for maintenance" in result: raise MaintenanceException("Server is down for maintenance") from error - raise Exception( + raise UnknownException( f"Unknown error while requesting {response.url}. {response.status} - {result}" ) from error @@ -823,7 +811,7 @@ async def check_response(response: ClientResponse) -> None: if "Not such token with UUID: " in message: raise NotSuchTokenException(message) - raise Exception(message if message else result) + raise UnknownException(message or result) async def _refresh_token_if_expired(self) -> None: """Check if token is expired and request a new one.""" diff --git a/pyoverkiz/exceptions.py b/pyoverkiz/exceptions.py index c85cfead..f2908b6e 100644 --- a/pyoverkiz/exceptions.py +++ b/pyoverkiz/exceptions.py @@ -79,3 +79,7 @@ class SomfyBadCredentialsException(BadCredentialsException): class SomfyServiceException(Exception): pass + + +class UnknownException(Exception): + pass diff --git a/pyoverkiz/models.py b/pyoverkiz/models.py index e2f79114..12c7ad96 100644 --- a/pyoverkiz/models.py +++ b/pyoverkiz/models.py @@ -316,10 +316,7 @@ class States: _states: list[State] def __init__(self, states: list[dict[str, Any]] | None = None) -> None: - if states: - self._states = [State(**state) for state in states] - else: - self._states = [] + self._states = [State(**state) for state in states] if states else [] def __iter__(self) -> Iterator[State]: return self._states.__iter__() diff --git a/pyoverkiz/obfuscate.py b/pyoverkiz/obfuscate.py index 59f897d8..11f84ea0 100644 --- a/pyoverkiz/obfuscate.py +++ b/pyoverkiz/obfuscate.py @@ -19,7 +19,7 @@ def obfuscate_email(email: str | None) -> str: def obfuscate_string(input: str) -> str: """Mask string""" - return re.sub(r"[a-zA-Z0-9_.-]*", "*", str(input)) + return re.sub(r"[a-zA-Z0-9_.-]*", "*", input) def obfuscate_sensitive_data(data: dict[str, Any]) -> JSON: