feat: Refactor http client to be async (#30)

* Refactor http client to be async

* Start testing http_client

* Make all client functions async

* Fix tests for http client

* Fix ping tests

* Fix read_subjects tests

* Fix read_subject tests

* Fix write events tests

* Fix read_events tests

* Fix observe event tests

* Refactor observe events tests

* Refactor read events tests

* Refactor tests

Author: dotKuro

eventsourcingdb_client_python/abstract_base_client.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from .http_client import HttpClient
+from .http_client.http_client import HttpClient
 
 
 class AbstractBaseClient(ABC):

eventsourcingdb_client_python/client.py

@@ -1,4 +1,4 @@
-from collections.abc import Generator
+from collections.abc import AsyncGenerator
 
 from .abstract_base_client import AbstractBaseClient
 from .client_configuration import ClientConfiguration
@@ -7,7 +7,7 @@
 from .event.event_context import EventContext
 from .handlers.observe_events.observe_events import observe_events
 from .handlers.observe_events.observe_events_options import ObserveEventsOptions
-from .http_client import HttpClient
+from .http_client.http_client import HttpClient
 from .handlers.ping import ping
 from .handlers.read_events import read_events, ReadEventsOptions
 from .handlers.read_subjects import read_subjects
@@ -16,54 +16,75 @@
 
 
 class Client(AbstractBaseClient):
-    def __init__(
-        self,
+    __create_key = object()
+
+    @classmethod
+    async def create(
+        cls,
         base_url: str,
         access_token: str,
         options: ClientOptions = ClientOptions()
-    ):
-        self.configuration: ClientConfiguration = ClientConfiguration(
+    ) -> 'Client':
+        configuration = ClientConfiguration(
             base_url=base_url,
             timeout_seconds=options.timeout_seconds,
             access_token=access_token,
             protocol_version=options.protocol_version,
             max_tries=options.max_tries
         )
 
-        self.__http_client: HttpClient = HttpClient(self.configuration)
+        http_client = await HttpClient.create(configuration)
+
+        return cls(Client.__create_key, http_client)
+
+    def __init__(
+        self,
+        create_key,
+        http_client: HttpClient
+    ):
+        assert create_key == Client.__create_key, \
+            'Client objects must be created using Client.create.'
+
+        self.__http_client = http_client
+
+    async def close(self):
+        await self.__http_client.close()
 
     @property
     def http_client(self) -> HttpClient:
         return self.__http_client
 
-    def ping(self) -> None:
-        return ping(self)
+    async def ping(self) -> None:
+        return await ping(self)
 
-    def read_subjects(
+    async def read_subjects(
         self,
         base_subject: str
-    ) -> Generator[str, None, None]:
-        return read_subjects(self, base_subject)
+    ) -> AsyncGenerator[str, None]:
+        async for subject in read_subjects(self, base_subject):
+            yield subject
 
-    def read_events(
+    async def read_events(
         self,
         subject: str,
         options: ReadEventsOptions
-    ) -> Generator[StoreItem, None, None]:
-        return read_events(self, subject, options)
+    ) -> AsyncGenerator[StoreItem, None]:
+        async for event in read_events(self, subject, options):
+            yield event
 
-    def observe_events(
+    async def observe_events(
         self,
         subject: str,
         options: ObserveEventsOptions
-    ) -> Generator[StoreItem, None, None]:
-        return observe_events(self, subject, options)
+    ) -> AsyncGenerator[StoreItem, None]:
+        async for event in observe_events(self, subject, options):
+            yield event
 
-    def write_events(
+    async def write_events(
         self,
         event_candidates: list[EventCandidate],
         preconditions: list[Precondition] = None
     ) -> list[EventContext]:
         if preconditions is None:
             preconditions = []
-        return write_events(self, event_candidates, preconditions)
+        return await write_events(self, event_candidates, preconditions)

eventsourcingdb_client_python/handlers/observe_events/observe_events.py

@@ -1,9 +1,7 @@
 import json
-from collections.abc import Generator
+from collections.abc import AsyncGenerator
 from http import HTTPStatus
 
-import requests
-
 from ..is_heartbeat import is_heartbeat
 from ..is_item import is_item
 from ..is_stream_error import is_stream_error
@@ -18,13 +16,14 @@
 from ...event.validate_subject import validate_subject
 from ..store_item import StoreItem
 from .observe_events_options import ObserveEventsOptions
+from ...http_client.response import Response
 
 
-def observe_events(
+async def observe_events(
     client: AbstractBaseClient,
     subject: str,
     options: ObserveEventsOptions
-) -> Generator[StoreItem, None, None]:
+) -> AsyncGenerator[StoreItem, None]:
     try:
         validate_subject(subject)
     except ValidationError as validation_error:
@@ -44,12 +43,11 @@ def observe_events(
         'options': options.to_json()
     })
 
-    response: requests.Response
+    response: Response
     try:
-        response = client.http_client.post(
+        response = await client.http_client.post(
             path='/api/observe-events',
             request_body=request_body,
-            stream_response=True
         )
     except CustomError as custom_error:
         raise custom_error
@@ -62,7 +60,7 @@ def observe_events(
                 f'Unexpected response status: '
                 f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
             )
-        for raw_message in response.iter_lines():
+        async for raw_message in response.body:
             message = parse_raw_message(raw_message)
 
             if is_heartbeat(message):

eventsourcingdb_client_python/handlers/ping.py

@@ -4,8 +4,9 @@
 from ..errors.server_error import ServerError
 
 
-def ping(client: AbstractBaseClient) -> None:
-    response = client.http_client.get('/ping')
+async def ping(client: AbstractBaseClient) -> None:
+    response = await client.http_client.get('/ping')
+    response_body = bytes.decode(await response.body.read(), encoding='utf-8')
 
-    if response.status_code != HTTPStatus.OK or response.text != HTTPStatus.OK.phrase:
-        raise ServerError(f'Received unexpected response: {response.text}')
+    if response.status_code != HTTPStatus.OK or response_body != HTTPStatus.OK.phrase:
+        raise ServerError(f'Received unexpected response: {response_body}')

eventsourcingdb_client_python/handlers/read_events/read_events.py

@@ -2,8 +2,6 @@
 import json
 from http import HTTPStatus
 
-import requests
-
 from ...abstract_base_client import AbstractBaseClient
 from ...errors.custom_error import CustomError
 from ...errors.internal_error import InternalError
@@ -12,14 +10,15 @@
 from ...errors.validation_error import ValidationError
 from ...event.event import Event
 from ...event.validate_subject import validate_subject
+from ...http_client.response import Response
 from ..is_item import is_item
 from ..is_stream_error import is_stream_error
 from ..parse_raw_message import parse_raw_message
 from ..store_item import StoreItem
 from .read_events_options import ReadEventsOptions
 
 
-def read_events(
+async def read_events(
     client: AbstractBaseClient,
     subject: str,
     options: ReadEventsOptions
@@ -43,12 +42,11 @@ def read_events(
         'options': options.to_json()
     })
 
-    response: requests.Response
+    response: Response
     try:
-        response = client.http_client.post(
+        response = await client.http_client.post(
             path='/api/read-events',
             request_body=request_body,
-            stream_response=True
         )
     except CustomError as custom_error:
         raise custom_error
@@ -61,7 +59,7 @@ def read_events(
                 f'Unexpected response status: '
                 f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
             )
-        for raw_message in response.iter_lines():
+        async for raw_message in response.body:
             message = parse_raw_message(raw_message)
 
             if is_stream_error(message):

eventsourcingdb_client_python/handlers/read_subjects/read_subjects.py

@@ -2,21 +2,20 @@
 import json
 from http import HTTPStatus
 
-import requests
-
 from ...abstract_base_client import AbstractBaseClient
 from ...errors.custom_error import CustomError
 from ...errors.internal_error import InternalError
 from ...errors.invalid_parameter_error import InvalidParameterError
 from ...errors.server_error import ServerError
 from ...errors.validation_error import ValidationError
+from ...http_client.response import Response
 from ..is_stream_error import is_stream_error
 from ..parse_raw_message import parse_raw_message
 from .is_subject import is_subject
 from ...event.validate_subject import validate_subject
 
 
-def read_subjects(
+async def read_subjects(
     client: AbstractBaseClient,
     base_subject: str
 ) -> Generator[str, None, None]:
@@ -31,12 +30,11 @@ def read_subjects(
         'baseSubject': base_subject
     })
 
-    response: requests.Response
+    response: Response
     try:
-        response = client.http_client.post(
+        response = await client.http_client.post(
             path='/api/read-subjects',
             request_body=request_body,
-            stream_response=True
         )
     except CustomError as custom_error:
         raise custom_error
@@ -49,7 +47,7 @@ def read_subjects(
                 f'Unexpected response status: '
                 f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
             )
-        for raw_message in response.iter_lines():
+        async for raw_message in response.body:
             message = parse_raw_message(raw_message)
 
             if is_stream_error(message):

eventsourcingdb_client_python/handlers/write_events/write_events.py

@@ -1,8 +1,5 @@
 from http import HTTPStatus
 import json
-from typing import Any
-
-import requests
 
 from ...abstract_base_client import AbstractBaseClient
 from ...errors.custom_error import CustomError
@@ -12,10 +9,11 @@
 from ...errors.validation_error import ValidationError
 from ...event.event_candidate import EventCandidate
 from ...event.event_context import EventContext
+from ...http_client.response import Response
 from .preconditions import Precondition
 
 
-def write_events(
+async def write_events(
     client: AbstractBaseClient,
     event_candidates: list[EventCandidate],
     preconditions: list[Precondition]
@@ -41,9 +39,9 @@ def write_events(
         'preconditions': [precondition.to_json() for precondition in preconditions]
     })
 
-    response: requests.Response
+    response: Response
     try:
-        response = client.http_client.post(
+        response = await client.http_client.post(
             path='/api/write-events',
             request_body=request_body,
         )
@@ -58,10 +56,11 @@ def write_events(
             f'{response.status_code} {HTTPStatus(response.status_code).phrase}.'
         )
 
-    response_data: Any
+    response_data = await response.body.read()
+    response_data = bytes.decode(response_data, encoding='utf-8')
     try:
-        response_data = response.json()
-    except requests.exceptions.JSONDecodeError as decode_error:
+        response_data = json.loads(response_data)
+    except json.JSONDecodeError as decode_error:
         raise ServerError(str(decode_error)) from decode_error
     except Exception as other_error:
         raise InternalError(str(other_error)) from other_error
@@ -70,13 +69,13 @@ def write_events(
         raise ServerError(
             f'Failed to parse response \'{response_data}\' to list.')
 
-    return_value = []
+    result = []
     for unparsed_event_context in response_data:
         try:
-            return_value.append(EventContext.parse(unparsed_event_context))
+            result.append(EventContext.parse(unparsed_event_context))
         except ValidationError as validation_error:
             raise ServerError(str(validation_error)) from validation_error
         except Exception as other_error:
             raise InternalError(str(other_error)) from other_error
 
-    return return_value
+    return result