fix PR comments and linting

Author: atstaeff

eventsourcingdb/client.py

@@ -129,9 +129,18 @@ def start(self) -> 'Container':
             return self
 
         try:
+            # Try to pull the latest image
             self._docker_client.images.pull(self._image_name, self._image_tag)
         except errors.APIError as e:
-            raise RuntimeError(f'Warning: Could not pull image: {e}') from e
+            # Check if we already have the image locally
+            try:
+                image_name = f"{self._image_name}:{self._image_tag}"
+                self._docker_client.images.get(image_name)
+                # If we get here, the image exists locally, so we can continue
+                logging.warning(f"Warning: Could not pull image: {e}. Using locally cached image.")
+            except errors.ImageNotFound:
+                # If the image isn't available locally either, we can't continue
+                raise RuntimeError(f'Could not pull image and no local image available: {e}') from e
 
         self._cleanup_existing_containers()
         self._create_container()

eventsourcingdb/container.py

@@ -9,7 +9,6 @@
 from .get_get_headers import get_get_headers
 from .get_post_headers import get_post_headers
 from .response import Response
-from .validate_response import validate_response
 
 
 class HttpClient:
@@ -58,14 +57,7 @@ async def __request_executor() -> Response:
 
             response = Response(async_response)
 
-            validated_response = None
-            try:
-                validated_response = await validate_response(response)
-            except Exception as error:
-                response.close()
-                raise error
-
-            return validated_response
+            return response
 
         return await __request_executor()
 
@@ -88,13 +80,6 @@ async def __request_executor() -> Response:
 
             response = Response(async_response)
 
-            validated_response = None
-            try:
-                validated_response = await validate_response(response)
-            except Exception as error:
-                response.close()
-                raise error
-
-            return validated_response
+            return response
 
         return await __request_executor()

eventsourcingdb/http_client/get_error_message.py

@@ -11,20 +11,20 @@ class Response:
     def __init__(self, response: aiohttp.ClientResponse):
         self.__response: aiohttp.ClientResponse = response
 
-    def __aenter__(self):
+    async def __aenter__(self):
+        # Properly await any async initialization if needed
         return self
 
-    def __aexit__(self, exc_type, exc_val, exc_tb):
-        self.close()
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        if not self.__response.closed:
+            self.__response.close()
 
     def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        self.close()
-
-    def close(self):
-        self.__response.close()
+        if not self.__response.closed:
+            self.__response.close()
 
     @property
     def status_code(self) -> HTTPStatus:
@@ -37,3 +37,6 @@ def headers(self) -> Headers:
     @property
     def body(self) -> StreamReader:
         return self.__response.content
+
+    def __str__(self) -> str:
+        return f'Response(status_code={self.status_code}, headers={dict(self.headers)})'

eventsourcingdb/http_client/http_client.py

@@ -0,0 +1,81 @@
+import asyncio
+import contextlib
+import signal
+from eventsourcingdb.client import Client
+from eventsourcingdb.handlers.read_events import ReadEventsOptions
+
+
+async def read_events_with_timeout(client, subject, timeout=5.0):
+    """Read events with a timeout, demonstrating cancellation."""
+    try:
+        # Use asyncio.wait_for to implement a timeout
+        await asyncio.wait_for(
+            process_events(client, subject),
+            timeout=timeout
+        )
+    except asyncio.TimeoutError:
+        print(f"Reading events timed out after {timeout} seconds")
+
+
+async def process_events(client, subject):
+    """Process events from the subject."""
+    print(f"Starting to read events from {subject}")
+    
+    # Using contextlib.aclosing ensures the generator is closed properly
+    # even if we exit the context due to an exception
+    async with contextlib.aclosing(
+        client.read_events(
+            subject=subject,
+            options=ReadEventsOptions(recursive=True)
+        )
+    ) as events:
+        # Process events until something stops the operation
+        count = 0
+        async for event in events:
+            count += 1
+            print(f"Processed event #{count}: {event.type}")
+            
+            # Example condition to stop processing
+            if count >= 10:
+                print("Reached maximum event count, stopping")
+                break
+                
+    print("Finished processing events")
+
+
+async def cancel_after_delay(task, delay=2.0):
+    """Cancel a task after a delay."""
+    await asyncio.sleep(delay)
+    print(f"Cancelling task after {delay} seconds")
+    task.cancel()
+
+
+async def main():
+    # Create client
+    client = Client(
+        base_url="http://localhost:8080",
+        api_token="your-api-token"
+    )
+    
+    try:
+        await client.initialize()
+        
+        # Example 1: Using asyncio.wait_for with a timeout
+        print("\n--- Example 1: Reading with timeout ---")
+        await read_events_with_timeout(client, "/users", timeout=3.0)
+        
+        # Example 2: Create a task and cancel it explicitly
+        print("\n--- Example 2: Manual cancellation ---")
+        event_task = asyncio.create_task(process_events(client, "/orders"))
+        cancel_task = asyncio.create_task(cancel_after_delay(event_task, 2.0))
+        
+        try:
+            await event_task
+        except asyncio.CancelledError:
+            print("Event reading was cancelled")
+        
+        await cancel_task
+        
+    finally:
+        await client.close()
+

eventsourcingdb/http_client/response.py

@@ -1,3 +1,3 @@
 [pytest]
-timeout = 20
+timeout = 30
 asyncio_default_fixture_loop_scope = function

eventsourcingdb/http_client/validate_response.py

@@ -1,33 +1,6 @@
 import pytest_asyncio
-from eventsourcingdb.client import Client
 from eventsourcingdb.event.event_candidate import EventCandidate
-from eventsourcingdb.http_client.http_client import HttpClient
 from .shared.database import Database
-from .shared.start_local_http_server import \
-    start_local_http_server, \
-    StopServer
-
-pytest_plugins = ('pytest_asyncio', )
-
-
-@pytest_asyncio.fixture
-async def get_http_client():
-    stop_server: StopServer | None = None
-    client: Client | None = None
-
-    async def getter(attach_handlers) -> HttpClient:
-        nonlocal stop_server
-        nonlocal client
-        client, stop_server = await start_local_http_server(attach_handlers)
-        return client.http_client
-
-    yield getter
-
-    if stop_server is not None:
-        stop_server()
-
-    if client is not None:
-        await client.close()
 
 
 @pytest_asyncio.fixture
@@ -39,26 +12,6 @@ async def database():
         await testing_db.stop()
 
 
-@pytest_asyncio.fixture
-async def get_client():
-    stop_server: StopServer | None = None
-    client: Client | None = None
-
-    async def getter(attach_handlers) -> Client:
-        nonlocal stop_server
-        nonlocal client
-        client, stop_server = await start_local_http_server(attach_handlers)
-        return client
-
-    yield getter
-
-    if stop_server is not None:
-        stop_server()
-
-    if client is not None:
-        await client.close()
-
-
 class TestData:
     TEST_SOURCE_STRING = 'tag:thenativeweb.io,2023:eventsourcingdb:test'
     REGISTERED_SUBJECT = '/users/registered'

examples/read_events_with_cancellation.py

@@ -11,11 +11,14 @@ class Database:
     __create_key = object()
     __container: Container
 
+    CLIENT_TYPE_WITH_AUTH = 'with_authorization'
+    CLIENT_TYPE_INVALID_URL = 'with_invalid_url'
+
     def __init__(
         self,
         create_key,
         with_authorization_client: Client,
-        with_invalid_url_client: Client,
+        with_invalid_url_client: Client ,
     ):
         assert create_key == Database.__create_key, \
             'Database objects must be created using Database.create.'
@@ -28,6 +31,8 @@ def __init__(
     @classmethod
     def _create_container(cls, api_token, image_tag):
         cls.__container = Container()
+        cls.__container.with_image_tag(image_tag)
+        cls.__container.with_api_token(api_token)
         return cls.__container
 
     @staticmethod
@@ -77,7 +82,13 @@ async def create(cls, max_retries=3, retry_delay=2.0) -> 'Database':
                 continue
 
             try:
-                with_authorization_client, with_invalid_url_client = await cls._initialize_clients(container, api_token)
+                (
+                    with_authorization_client,
+                    with_invalid_url_client
+                ) = await cls._initialize_clients(
+                    container,
+                    api_token
+                )
             except Exception as client_error:
                 container.stop()
                 raise client_error
@@ -95,13 +106,15 @@ def _get_image_tag_from_dockerfile():
             content = dockerfile.read().strip()
             return content.split(':')[-1]
 
-    def get_client(self, client_type: str = "with_authorization") -> Client:
-        if client_type == "with_authorization":
+    def get_client(self, client_type: str = CLIENT_TYPE_WITH_AUTH) -> Client:
+        if client_type == self.CLIENT_TYPE_WITH_AUTH:
             return self.__with_authorization_client
-        elif client_type == "with_invalid_url":
+        if client_type == self.CLIENT_TYPE_INVALID_URL:
             return self.__with_invalid_url_client
-        else:
-            raise ValueError(f"Unknown client type: {client_type}")
+
+        raise ValueError(f"Unknown client type: {client_type}")
 
     async def stop(self) -> None:
-        self.__class__.__container.stop()
+        # Use walrus operator for concise access and check
+        if (container := getattr(self.__class__, '_Database__container', None)):
+            container.stop()