fix httpclient

Author: atstaeff

eventsourcingdb/client.py

@@ -1,4 +1,6 @@
 from collections.abc import AsyncGenerator
+from types import TracebackType
+from typing import Optional, Type, TypeVar, AsyncIterator
 
 from .client_configuration import ClientConfiguration
 from .event.event_candidate import EventCandidate
@@ -16,34 +18,36 @@
 from .handlers.write_events import Precondition, write_events
 
 
-# pylint: disable=R6007
-# Reason: This class explicitly specifies the return type as None
-# for better readability. Even though it is not necessary,
-# it makes the return type clear without needing to read any
-# documentation or code.
-class Client():
+T = TypeVar('T')
+
+class Client:
     def __init__(
         self,
         base_url: str,
         api_token: str,
     ):
-        configuration = ClientConfiguration(
-            base_url=base_url,
-            api_token=api_token,
-        )
+        self.__http_client = HttpClient(base_url=base_url, api_token=api_token)
 
-        self.__http_client = HttpClient(configuration)
+    async def __aenter__(self):
+        await self.__http_client.initialize()
+        return self
+
+    async def __aexit__(
+        self, 
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType]
+    ) -> None:
+        await self.__http_client.close()
 
-    #TODO: is this necessary? __enter__, __exit__
-    async def initialize(self) -> None:
+    # Keeping these for backward compatibility and explicit resource management
+    """async def initialize(self) -> None:
         await self.__http_client.initialize()
 
-     #TODO: is this necessary? magic method __enter__, __exit__
     async def close(self) -> None:
-        await self.__http_client.close()
-
-    @property
-    def http_client(self) -> HttpClient:
+        await self.__http_client.close() # TODO: should we mix object orientation and functional programming?
+    """
+    def http_client(self) -> # TODO: should we mix object orientation and functional programming?tpClient:
         return self.__http_client
 
     # TODO: should we mix object orientation and functional programming?
@@ -89,9 +93,13 @@ async def read_events(
             return http.status(200);
         }
 """
-        async for event in read_events(self, subject, options):
-            yield event
-
+        """Read events with proper cancellation support."""
+        generator = read_events(self, subject, options)
+        try:
+            async for item in generator:
+                yield item
+        finally:
+            await generator.aclose()
     # TODO: run eventql query
     async def run_eventql_query(self, query: str) -> AsyncGenerator[Event, None]:
         """
@@ -106,26 +114,42 @@ async def observe_events(
         """
         TODO: the same issue like read_events. contextmanager
         """
-        async for event in observe_events(self, subject, options):
-            yield event
-
-    async def register_event_schema(self, event_type: str, json_schema: str) -> None: # TODO: no json_schema is dict no string anymore
-        # no context manager liek read_events
+    async def observe_events(
+        self,
+        subject: str,
+        options: ObserveEventsOptions
+    ) -> AsyncGenerator[StoreItem, None]:
+        generator = observe_events(self, subject, options)
+        try:
+            async for item in generator:
+                yield item
+        finally:
+            await generator.aclose()
+
+    async def register_event_schema(self, event_type: str, json_schema: dict) -> None:
+        # Updated type hint to reflect it should be a dict
         await register_event_schema(self, event_type, json_schema)
 
     async def read_subjects(
         self,
         base_subject: str
     ) -> AsyncGenerator[str, None]:
-        # TODO: the same issue like read_events. contextmanager
-        async for subject in read_subjects(self, base_subject):
-            yield subject
-
+        """Read subjects with proper cancellation support."""
+        generator = read_subjects(self, base_subject)
+        try:
+            async for item in generator:
+                yield item
+        finally:
+            await generator.aclose()
 
     async def read_event_types(self) -> AsyncGenerator[EventType, None]:
-        # TODO: the same issue like read_events. contextmanager
-        async for event_type in read_event_types(self):
-            yield event_type
+        """Read event types with proper cancellation support."""
+        generator = read_event_types(self)
+        try:
+            async for item in generator:
+                yield item
+        finally:
+            await generator.aclose()
 
 
 

eventsourcingdb/http_client/http_client.py

@@ -1,20 +1,17 @@
 from collections import namedtuple
 from collections.abc import Callable, Coroutine
-from http import HTTPStatus
+from types import TracebackType
+from typing import Any, Optional, Type
 
 import aiohttp
 from aiohttp import ClientSession
 
-from .response import Response, Headers
-from ..client_configuration import ClientConfiguration
+from .response import Response
 from ..errors.client_error import ClientError
 from ..errors.custom_error import CustomError
 from ..errors.internal_error import InternalError
 from ..errors.server_error import ServerError
 from ..util import url
-from ..util.retry.retry_result import Retry, Return, RetryResult
-from ..util.retry.retry_with_backoff import retry_with_backoff
-from ..util.retry.retry_error import RetryError
 
 Range = namedtuple('Range', 'lower, upper')
 
@@ -26,71 +23,62 @@ class UninitializedError(CustomError):
 class HttpClient:
     def __init__(
         self,
-        client_configuration: ClientConfiguration
+        base_url: str,
+        api_token: str,
     ):
-        self.__client_configuration: ClientConfiguration = client_configuration
+        self.__base_url = base_url
+        self.__api_token = api_token
         self.__session: ClientSession | None = None
+    
+    async def __aenter__(self):
+        await self.initialize()
+        return self
+    
+    async def __aexit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType]
+    ) -> None:
+        await self.close()
 
+    # Keep for backward compatibility
     async def initialize(self) -> None:
-        self.__session = aiohttp.ClientSession(
-            timeout=aiohttp.ClientTimeout(
-                connect=self.__client_configuration.timeout_seconds,
-                sock_read=self.__client_configuration.timeout_seconds
-            )
-        )
+        self.__session = aiohttp.ClientSession()
 
+    # Keep for backward compatibility
     async def close(self):
         if self.__session is not None:
             await self.__session.close()
+            self.__session = None
 
     async def __execute_request(
         self,
-        execute_request: Callable[[], Coroutine[None, None, RetryResult[Response]]]
+        execute_request: Callable[[], Coroutine[Any, Any, Response]]
     ) -> Response:
         try:
-            return await retry_with_backoff(
-                self.__client_configuration.max_tries,
-                execute_request
-            )
-        except RetryError as retry_error:
-            raise ServerError(str(retry_error)) from retry_error
+            result = await execute_request()
+            return result
         except CustomError as custom_error:
             raise custom_error
         except aiohttp.ClientError as request_error:
             raise ServerError(str(request_error)) from request_error
         except Exception as other_error:
             raise InternalError(str(other_error)) from other_error
 
-    def __validate_protocol_version(self, http_status_code: int, headers: Headers) -> None:
-        if http_status_code != HTTPStatus.UNPROCESSABLE_ENTITY:
-            return
-
-        server_protocol_version = headers.get(
-            'x-eventsourcingdb-protocol-version'
-        )
-
-        if server_protocol_version is None:
-            server_protocol_version = 'unknown version'
-
-        raise ClientError(
-            f'Protocol version mismatch, server \'{server_protocol_version}\','
-            f' client \'{self.__client_configuration.protocol_version}\'.'
-        )
 
     @staticmethod
     async def __get_error_message(response: Response):
         error_message = f'Request failed with status code \'{response.status_code}\''
 
         # We want to purposefully ignore all errors here, as we're already error handling,
         # and this function just tries to get more information on a best-effort basis.
-        # pylint: disable=too-many-try-statements
         try:
             encoded_error_reason = await response.body.read()
             error_reason = encoded_error_reason.decode('utf-8')
             error_message += f" {error_reason}"
         finally:
             pass
-        # pylint: enable=too-many-try-statements
 
         error_message += '.'
 
@@ -99,23 +87,20 @@ async def __get_error_message(response: Response):
     async def __validate_response(
         self,
         response: Response
-    ) -> RetryResult[Response]:
+    ) -> Response:
         server_failure_range = Range(500, 600)
         if server_failure_range.lower <= response.status_code < server_failure_range.upper:
-            return Retry(ServerError(await self.__get_error_message(response)))
+            raise ServerError(await self.__get_error_message(response))
 
         client_failure_range = Range(400, 500)
         if client_failure_range.lower <= response.status_code < client_failure_range.upper:
             raise ClientError(await self.__get_error_message(response))
 
-        self.__validate_protocol_version(response.status_code, response.headers)
-
-        return Return(response)
+        return response
 
     def __get_post_request_headers(self) -> dict[str, str]:
         headers = {
-            'X-EventSourcingDB-Protocol-Version': self.__client_configuration.protocol_version,
-            'Authorization': f'Bearer {self.__client_configuration.api_token}',
+            'Authorization': f'Bearer {self.__api_token}',
             'Content-Type': 'application/json'
         }
 
@@ -125,37 +110,31 @@ async def post(self, path: str, request_body: str) -> Response:
         if self.__session is None:
             raise UninitializedError()
 
-        async def execute_request() -> RetryResult[Response]:
-            response = await self.__session.post(
+        async def execute_request() -> Response:
+            async_response = await self.__session.post( # type: ignore
                 url.join_segments(
-                    self.__client_configuration.base_url,
+                    self.__base_url,
                     path
                 ),
                 data=request_body,
                 headers=self.__get_post_request_headers(),
             )
 
-            response = Response(response)
+            response = Response(async_response)
             try:
                 result = await self.__validate_response(response)
+                return result
             except Exception as error:
                 response.close()
                 raise error
 
-            if isinstance(result, Retry):
-                response.close()
-
-            return result
-
         return await self.__execute_request(execute_request)
 
     def __get_get_request_headers(self, with_authorization: bool) -> dict[str, str]:
-        headers = {
-            'X-EventSourcingDB-Protocol-Version': self.__client_configuration.protocol_version,
-        }
+        headers = {}
 
         if with_authorization:
-            headers['Authorization'] = f'Bearer {self.__client_configuration.api_token}'
+            headers['Authorization'] = f'Bearer {self.__api_token}'
 
         return headers
 
@@ -167,23 +146,21 @@ async def get(
         if self.__session is None:
             raise UninitializedError()
 
-        async def execute_request() -> RetryResult[Response]:
-            response = await self.__session.get(
+        async def execute_request() -> Response:
+            async_response = await self.__session.get( # type: ignore
                 url.join_segments(
-                    self.__client_configuration.base_url, path),
+                    self.__base_url, 
+                    path
+                ),
                 headers=self.__get_get_request_headers(with_authorization),
             )
 
-            response = Response(response)
+            response = Response(async_response)
             try:
                 result = await self.__validate_response(response)
+                return result
             except Exception as error:
                 response.close()
                 raise error
 
-            if isinstance(result, Retry):
-                response.close()
-
-            return result
-
         return await self.__execute_request(execute_request)

eventsourcingdb/http_client/response.py

@@ -3,6 +3,7 @@
 from http import HTTPStatus
 
 import aiohttp
+from aiohttp import streams as aiohttp_streams
 
 Headers = Mapping[str, str]
 
@@ -11,10 +12,10 @@ class Response:
     def __init__(self, response: aiohttp.ClientResponse):
         self.__response: aiohttp.ClientResponse = response
 
-    def __enter__(self):
+    def __aenter__(self):
         return self
 
-    def __exit__(self, exc_type, exc_val, exc_tb):
+    def __aexit__(self, exc_type, exc_val, exc_tb):
         self.close()
 
     def close(self):
@@ -27,7 +28,6 @@ def status_code(self) -> HTTPStatus:
     @property
     def headers(self) -> Headers:
         return self.__response.headers
-
     @property
-    def body(self) -> asyncio.StreamReader:
+    def body(self) -> aiohttp_streams.StreamReader:
         return self.__response.content