Run ruff formatter

Author: nanomad

examples/main.py

@@ -7,10 +7,11 @@
 
 
 async def main():
-    config = SaicApiConfiguration(username="[email protected]", password="XXXXX", )
-    saic_api = SaicApi(
-        config
+    config = SaicApiConfiguration(
+        username="[email protected]",
+        password="XXXXX",
     )
+    saic_api = SaicApi(config)
     await saic_api.login()
     while True:
         logging.info("Auth token expires at %s", saic_api.token_expiration)
@@ -26,7 +27,7 @@ async def main():
         await asyncio.sleep(10)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     logging.basicConfig(
         stream=sys.stdout,
         format="%(levelname)s [%(asctime)s] %(name)s - %(message)s",

src/saic_ismart_client_ng/__init__.py

@@ -15,6 +15,6 @@ class SaicApi(
     SaicVehicleLocksApi,
     SaicVehicleWindowsApi,
     SaicVehicleClimateApi,
-    SaicVehicleChargingApi
+    SaicVehicleChargingApi,
 ):
-    """ The SAIC Api client """
+    """The SAIC Api client"""

src/saic_ismart_client_ng/api/base.py

@@ -12,21 +12,25 @@
 
 from saic_ismart_client_ng.api.schema import LoginResp
 from saic_ismart_client_ng.crypto_utils import sha1_hex_digest
-from saic_ismart_client_ng.exceptions import SaicApiException, SaicApiRetryException, SaicLogoutException
+from saic_ismart_client_ng.exceptions import (
+    SaicApiException,
+    SaicApiRetryException,
+    SaicLogoutException,
+)
 from saic_ismart_client_ng.listener import SaicApiListener
 from saic_ismart_client_ng.model import SaicApiConfiguration
 from saic_ismart_client_ng.net.client import SaicApiClient
 
 logger = logging.getLogger(__name__)
 
-T = TypeVar('T')
+T = TypeVar("T")
 
 
 class AbstractSaicApi(ABC):
     def __init__(
-            self,
-            configuration: SaicApiConfiguration,
-            listener: Optional[SaicApiListener] = None,
+        self,
+        configuration: SaicApiConfiguration,
+        listener: Optional[SaicApiListener] = None,
     ):
         self.__configuration = configuration
         self.__api_client = SaicApiClient(configuration, listener=listener)
@@ -36,7 +40,7 @@ async def login(self) -> LoginResp:
         headers = {
             "Content-Type": "application/x-www-form-urlencoded",
             "Accept": "application/json",
-            "Authorization": "Basic c3dvcmQ6c3dvcmRfc2VjcmV0"
+            "Authorization": "Basic c3dvcmQ6c3dvcmRfc2VjcmV0",
         }
         firebase_device_id = "cqSHOMG1SmK4k-fzAeK6hr:APA91bGtGihOG5SEQ9hPx3Dtr9o9mQguNiKZrQzboa-1C_UBlRZYdFcMmdfLvh9Q_xA8A0dGFIjkMhZbdIXOYnKfHCeWafAfLXOrxBS3N18T4Slr-x9qpV6FHLMhE9s7I6s89k9lU7DD"
         form_body = {
@@ -47,32 +51,36 @@ async def login(self) -> LoginResp:
             "deviceId": f"{firebase_device_id}###europecar",
             "deviceType": "1",  # 2 for huawei
             "loginType": "2" if self.__configuration.username_is_email else "1",
-            "countryCode": "" if self.__configuration.username_is_email else self.__configuration.phone_country_code,
+            "countryCode": ""
+            if self.__configuration.username_is_email
+            else self.__configuration.phone_country_code,
         }
 
         result = await self.execute_api_call(
             "POST",
             "/oauth/token",
             form_body=form_body,
             out_type=LoginResp,
-            headers=headers
+            headers=headers,
         )
         # Update the user token
         self.__api_client.user_token = result.access_token
-        self.__token_expiration = datetime.datetime.now() + datetime.timedelta(seconds=result.expires_in)
+        self.__token_expiration = datetime.datetime.now() + datetime.timedelta(
+            seconds=result.expires_in
+        )
         return result
 
     async def execute_api_call(
-            self,
-            method: str,
-            path: str,
-            *,
-            body: Optional[Any] = None,
-            form_body: Optional[Any] = None,
-            out_type: Optional[Type[T]] = None,
-            params: Optional[QueryParamTypes] = None,
-            headers: Optional[HeaderTypes] = None,
-            allow_null_body: bool = False,
+        self,
+        method: str,
+        path: str,
+        *,
+        body: Optional[Any] = None,
+        form_body: Optional[Any] = None,
+        out_type: Optional[Type[T]] = None,
+        params: Optional[QueryParamTypes] = None,
+        headers: Optional[HeaderTypes] = None,
+        allow_null_body: bool = False,
     ) -> T:
         try:
             return await self.__execute_api_call(
@@ -83,41 +91,45 @@ async def execute_api_call(
                 out_type=out_type,
                 params=params,
                 headers=headers,
-                allow_null_body=allow_null_body
+                allow_null_body=allow_null_body,
             )
         except SaicApiException as e:
             raise e
         except Exception as e:
-            raise SaicApiException(f"API call {method} {path} failed unexpectedly", return_code=500) from e
+            raise SaicApiException(
+                f"API call {method} {path} failed unexpectedly", return_code=500
+            ) from e
 
     async def __execute_api_call(
-            self,
-            method: str,
-            path: str,
-            *,
-            body: Optional[Any] = None,
-            form_body: Optional[Any] = None,
-            out_type: Optional[Type[T]] = None,
-            params: Optional[QueryParamTypes] = None,
-            headers: Optional[HeaderTypes] = None,
-            allow_null_body: bool = False,
+        self,
+        method: str,
+        path: str,
+        *,
+        body: Optional[Any] = None,
+        form_body: Optional[Any] = None,
+        out_type: Optional[Type[T]] = None,
+        params: Optional[QueryParamTypes] = None,
+        headers: Optional[HeaderTypes] = None,
+        allow_null_body: bool = False,
     ) -> Optional[T]:
         url = f"{self.__configuration.base_uri}{path[1:] if path.startswith('/') else path}"
         json_body = asdict(body) if body else None
-        req = httpx.Request(method, url, params=params, headers=headers, data=form_body, json=json_body)
+        req = httpx.Request(
+            method, url, params=params, headers=headers, data=form_body, json=json_body
+        )
         response = await self.__api_client.send(req)
         return await self.__deserialize(req, response, out_type, allow_null_body)
 
     async def execute_api_call_with_event_id(
-            self,
-            method: str,
-            path: str,
-            *,
-            body: Optional[Any] = None,
-            out_type: Optional[Type[T]] = None,
-            params: Optional[QueryParamTypes] = None,
-            headers: Optional[HeaderTypes] = None,
-            delay: Optional[tenacity.wait.WaitBaseT] = None,
+        self,
+        method: str,
+        path: str,
+        *,
+        body: Optional[Any] = None,
+        out_type: Optional[Type[T]] = None,
+        params: Optional[QueryParamTypes] = None,
+        headers: Optional[HeaderTypes] = None,
+        delay: Optional[tenacity.wait.WaitBaseT] = None,
     ) -> Optional[T]:
         @tenacity.retry(
             stop=tenacity.stop_after_delay(30),
@@ -128,55 +140,60 @@ async def execute_api_call_with_event_id(
         )
         async def execute_api_call_with_event_id_inner(*, event_id: str):
             actual_headers = headers or dict()
-            actual_headers.update({'event-id': event_id})
+            actual_headers.update({"event-id": event_id})
             return await self.__execute_api_call(
                 method,
                 path,
                 body=body,
                 out_type=out_type,
                 params=params,
-                headers=actual_headers
+                headers=actual_headers,
             )
 
-        return await execute_api_call_with_event_id_inner(event_id='0')
+        return await execute_api_call_with_event_id_inner(event_id="0")
 
     async def __deserialize(
-            self,
-            request: httpx.Request,
-            response: httpx.Response,
-            data_class: Optional[Type[T]],
-            allow_null_body: bool
+        self,
+        request: httpx.Request,
+        response: httpx.Response,
+        data_class: Optional[Type[T]],
+        allow_null_body: bool,
     ) -> Optional[T]:
-
         try:
-            request_event_id = request.headers.get('event-id')
+            request_event_id = request.headers.get("event-id")
             json_data = response.json()
-            return_code = json_data.get('code', -1)
-            error_message = json_data.get('message', 'Unknown error')
+            return_code = json_data.get("code", -1)
+            error_message = json_data.get("message", "Unknown error")
             logger.debug(f"Response code: {return_code} {response.text}")
 
             if return_code in (401, 403) or response.status_code in (401, 403):
                 self.logout()
                 raise SaicLogoutException(response.text, return_code)
 
             if return_code in (2, 3, 7):
-                logger.error(f"API call return code is not acceptable: {return_code}: {response.text}")
+                logger.error(
+                    f"API call return code is not acceptable: {return_code}: {response.text}"
+                )
                 raise SaicApiException(error_message, return_code=return_code)
 
-            if 'event-id' in response.headers and 'data' not in json_data:
-                event_id = response.headers['event-id']
-                logger.info(f"Retrying since we got event-id in headers: {event_id}, but no data")
-                raise SaicApiRetryException(error_message, event_id=event_id, return_code=return_code)
+            if "event-id" in response.headers and "data" not in json_data:
+                event_id = response.headers["event-id"]
+                logger.info(
+                    f"Retrying since we got event-id in headers: {event_id}, but no data"
+                )
+                raise SaicApiRetryException(
+                    error_message, event_id=event_id, return_code=return_code
+                )
 
             if return_code != 0:
-                if request_event_id is not None and request_event_id != '0':
+                if request_event_id is not None and request_event_id != "0":
                     logger.info(
                         f"API call asked us to retry: {return_code}: {response.text}. Event id was: {request_event_id}"
                     )
                     raise SaicApiRetryException(
                         error_message,
                         event_id=request_event_id,
-                        return_code=return_code
+                        return_code=return_code,
                     )
                 else:
                     logger.error(
@@ -186,17 +203,19 @@ async def __deserialize(
 
             if data_class is None:
                 return None
-            elif 'data' in json_data:
+            elif "data" in json_data:
                 if data_class is str:
-                    return json.dumps(json_data['data'])
+                    return json.dumps(json_data["data"])
                 elif data_class is dict:
-                    return json_data['data']
+                    return json_data["data"]
                 else:
-                    return dacite.from_dict(data_class, json_data['data'])
+                    return dacite.from_dict(data_class, json_data["data"])
             elif allow_null_body:
                 return None
             else:
-                raise SaicApiException(f"Failed to deserialize response, missing 'data' field: {response.text}")
+                raise SaicApiException(
+                    f"Failed to deserialize response, missing 'data' field: {response.text}"
+                )
 
         except SaicApiException as se:
             raise se
@@ -205,29 +224,33 @@ async def __deserialize(
                 if response.status_code in (401, 403):
                     logger.error(
                         f"API call failed due to an authentication failure: {response.status_code} {response.text}",
-                        exc_info=e
+                        exc_info=e,
                     )
                     self.logout()
-                    raise SaicLogoutException(response.text, response.status_code) from e
+                    raise SaicLogoutException(
+                        response.text, response.status_code
+                    ) from e
                 else:
                     logger.error(
                         f"API call failed: {response.status_code} {response.text}",
-                        exc_info=e
+                        exc_info=e,
                     )
                     raise SaicApiException(response.text, response.status_code) from e
             else:
-                raise SaicApiException(f"Failed to deserialize response: {e}. Original json was {response.text}") from e
+                raise SaicApiException(
+                    f"Failed to deserialize response: {e}. Original json was {response.text}"
+                ) from e
 
     def logout(self):
-        self.__api_client.user_token = ''
+        self.__api_client.user_token = ""
         self.__token_expiration = None
 
     @property
     def is_logged_in(self) -> bool:
         return (
-                self.__api_client.user_token is not None and
-                self.__token_expiration is not None and
-                self.__token_expiration > datetime.datetime.now()
+            self.__api_client.user_token is not None
+            and self.__token_expiration is not None
+            and self.__token_expiration > datetime.datetime.now()
         )
 
     @property
@@ -238,9 +261,11 @@ def token_expiration(self) -> Optional[datetime.datetime]:
 def saic_api_after_retry(retry_state):
     wrapped_exception = retry_state.outcome.exception()
     if isinstance(wrapped_exception, SaicApiRetryException):
-        if 'event_id' in retry_state.kwargs:
-            logger.debug(f"Updating event_id to the newly obtained value {wrapped_exception.event_id}")
-            retry_state.kwargs['event_id'] = wrapped_exception.event_id
+        if "event_id" in retry_state.kwargs:
+            logger.debug(
+                f"Updating event_id to the newly obtained value {wrapped_exception.event_id}"
+            )
+            retry_state.kwargs["event_id"] = wrapped_exception.event_id
         else:
             logger.debug("Retrying without an event_id")