feat: BaseClient to share methods between Client & ClientAsync

Author: foarsitter

src/checkedid/client.py

@@ -19,22 +19,21 @@
 _T = TypeVar("_T")
 
 
-class Client:
+class BaseClient:
     ERROR_RESPONSE_MAPPING: Dict[int, Type[errors.CheckedIDError]] = {
         422: errors.CheckedIDValidationError,
         403: errors.CheckedIDAuthenticationError,
         404: errors.CheckedIDNotFoundError,
     }
 
     def __init__(self, customer_code: str, base_url: str = "https://api.checkedid.eu/"):
-        self.client: Optional[httpx.Client] = None
         self.base_url = base_url
-        self.create_client(base_url)
         self.access_token: Optional[str] = None
         self.customer_code = customer_code
+        self.create_client(base_url)
 
     def create_client(self, base_url: URLTypes) -> None:
-        self.httpx = httpx.Client(base_url=base_url, auth=self.authenticate_request)
+        raise NotImplementedError
 
     def authenticate_request(self, request: Request) -> Request:
         if self.access_token:
@@ -51,10 +50,38 @@ def process_response(
 
         return None
 
+    def handle_error_response(self, response: Response) -> None:
+        if response.status_code == 400:
+            raise errors.CheckedIDValidationError(
+                response.text, status_code=response.status_code
+            )
+
+        try:
+            json = response.json()
+        except JSONDecodeError:
+            json = {"message": response.text}
+
+        json["status_code"] = response.status_code
+
+        exception_type = self.map_exception(response)
+        raise exception_type(
+            status_code=response.status_code, json=json, message="Error from server"
+        )
+
+    def map_exception(self, response: Response) -> Type[errors.CheckedIDError]:
+        exception_type = self.ERROR_RESPONSE_MAPPING.get(
+            response.status_code, errors.CheckedIDError
+        )
+        return exception_type
+
+
+class Client(BaseClient):
+    client: httpx.Client
+
     def oauth_token(
         self, grant_type: str, username: str, password: str
     ) -> Optional[models.OAuthToken]:
-        response = self.httpx.post(
+        response = self.client.post(
             "/oauth/token",
             data={"grant_type": grant_type, "username": username, "password": password},
         )
@@ -67,8 +94,14 @@ def oauth_token(
             return typed_response
         return None
 
+    def __init__(self, customer_code: str, base_url: str = "https://api.checkedid.eu/"):
+        super().__init__(customer_code, base_url)
+
+    def create_client(self, base_url: URLTypes) -> None:
+        self.client = httpx.Client(base_url=base_url, auth=self.authenticate_request)
+
     def invitation_status(self, invitation_code: str) -> Optional[models.Invitation]:
-        response: Response = self.httpx.get(
+        response: Response = self.client.get(
             f"/result/status/{invitation_code}",
             headers={"Accept": "application/json"},
         )
@@ -82,7 +115,7 @@ def invitations_create(
             CustomerCode=self.customer_code, Invitations=invitations
         )
 
-        response: Response = self.httpx.post(
+        response: Response = self.client.post(
             "/invitations",
             json=obj.dict(),
             headers={"Accept": "application/json", "Content-Type": "application/json"},
@@ -91,7 +124,7 @@ def invitations_create(
         return self.process_response(response, models.CustomerDetails)
 
     def invitation_delete(self, invitation_code: str) -> bool:
-        response: Response = self.httpx.delete(
+        response: Response = self.client.delete(
             f"/invitation/{self.customer_code}/{invitation_code}",
             headers={"Accept": "application/json"},
         )
@@ -104,60 +137,55 @@ def invitation_delete(self, invitation_code: str) -> bool:
         return False
 
     def dossier(self, dossier_number: str) -> Optional[models.ReportResponse]:
-        response = self.httpx.get(f"/report/{dossier_number}")
+        response = self.client.get(f"/report/{dossier_number}")
 
         return self.process_response(response, models.ReportResponse)
 
     def dossier_with_scope(
         self, dossier_number: str, scope: str
     ) -> Optional[models.ReportDataV3]:
-        response = self.httpx.get(f"/reportdata/{dossier_number}/{scope}")
+        response = self.client.get(f"/reportdata/{dossier_number}/{scope}")
 
         return self.process_response(response, models.ReportDataV3)
 
-    def handle_error_response(self, response: Response) -> None:
-        if response.status_code == 400:
-            raise errors.CheckedIDValidationError(
-                response.text, status_code=response.status_code
-            )
 
-        try:
-            json = response.json()
-        except JSONDecodeError:
-            json = {"message": response.text}
+class ClientAsync(BaseClient):
+    """for asyncio"""
 
-        json["status_code"] = response.status_code
+    client: httpx.AsyncClient
 
-        exception_type = self.map_exception(response)
-        raise exception_type(
-            status_code=response.status_code, json=json, message="Error from server"
-        )
+    def __init__(self, customer_code: str, base_url: str = "https://api.checkedid.eu/"):
+        super().__init__(customer_code, base_url)
 
-    def map_exception(self, response: Response) -> Type[errors.CheckedIDError]:
-        exception_type = self.ERROR_RESPONSE_MAPPING.get(
-            response.status_code, errors.CheckedIDError
+    async def oauth_token(
+        self, grant_type: str, username: str, password: str
+    ) -> Optional[models.OAuthToken]:
+        response = await self.client.post(
+            "/oauth/token",
+            data={"grant_type": grant_type, "username": username, "password": password},
         )
-        return exception_type
 
+        typed_response = self.process_response(response, models.OAuthToken)
 
-class ClientAsync(Client):
-    """for asyncio"""
+        if typed_response:
+            self.access_token = typed_response.access_token
 
-    aclient: httpx.AsyncClient
+            return typed_response
+        return None
 
     def create_client(self, base_url: URLTypes) -> None:
-        self.aclient = httpx.AsyncClient(base_url=base_url)
+        self.client = httpx.AsyncClient(base_url=base_url)
 
-    async def adossier(self, dossier_number: str) -> Optional[models.ReportResponse]:
-        response = await self.aclient.get(
+    async def dossier(self, dossier_number: str) -> Optional[models.ReportResponse]:
+        response = await self.client.get(
             url=endpoints.DossierEndpoint.url(dossier_number=dossier_number)
         )
 
         return self.process_response(response, endpoints.DossierEndpoint.response)
 
     async def close(self) -> None:
-        if self.aclient:
-            await self.aclient.aclose()
+        if self.client:
+            await self.client.aclose()
 
     def open(self) -> None:
         self.create_client(self.base_url)

tests/test_async_dossiers.py

@@ -8,6 +8,6 @@ async def test_dossiers(customer_code, dossier_response_200):
     client = ClientAsync(customer_code)
 
     with client as client:
-        response = await client.dossier("999999-8888800")
+        response = await client.adossier("999999-8888800")
 
         assert response.DossierNumber == "999999-8888800"