feat: modify the schema by using item. Remove the function is_item. Not needed anymore. Payload has changed.

Author: atstaeff

eventsourcingdb_client_python/handlers/is_event.py

@@ -0,0 +1,21 @@
+from typing import Any
+
+EVENT_TYPE = 'event'
+
+
+def is_event(message: Any) -> bool:
+    # Check if this is the new format message
+    if isinstance(message, dict) and message.get('type') == EVENT_TYPE:
+        payload = message.get('payload')
+        return isinstance(payload, dict) and isinstance(payload.get('hash'), str)
+        
+    # Otherwise check for old format (just for backward compatibility)
+    if not isinstance(message, dict):
+        return False
+        
+    payload = message.get('payload')
+    return (
+        isinstance(payload, dict)
+        and isinstance(payload.get('event'), dict)
+        and isinstance(payload.get('hash'), str)
+    )

eventsourcingdb_client_python/handlers/is_item.py

@@ -3,7 +3,7 @@
 from http import HTTPStatus
 
 from ..is_heartbeat import is_heartbeat
-from ..is_item import is_item
+from ..is_event import is_event
 from ..is_stream_error import is_stream_error
 from ..parse_raw_message import parse_raw_message
 from ...abstract_base_client import AbstractBaseClient
@@ -73,8 +73,8 @@ async def observe_events(
             if is_stream_error(message):
                 raise ServerError(f'{message["payload"]["error"]}.')
 
-            if is_item(message):
-                event = Event.parse(message['payload']['event'])
+            if is_event(message):
+                event = Event.parse(message['payload'])
 
                 yield StoreItem(event, message['payload']['hash'])
                 continue

eventsourcingdb_client_python/handlers/observe_events/observe_events.py

@@ -11,7 +11,7 @@
 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_event import is_event
 from ..is_stream_error import is_stream_error
 from ..parse_raw_message import parse_raw_message
 from ..store_item import StoreItem
@@ -69,10 +69,10 @@ async def read_events(
             if is_stream_error(message):
                 raise ServerError(f'{message["payload"]["error"]}.')
 
-            if is_item(message):
+            if is_event(message):
                 event = Event.parse(message['payload']['event'])
 
-                yield StoreItem(event, message['payload']['hash'])
+                yield StoreItem(event, message['payload']['hash']) # type: ignore
                 continue
 
             raise ServerError(

eventsourcingdb_client_python/handlers/read_events/read_events.py

@@ -1,9 +1,8 @@
 from dataclasses import dataclass
 import subprocess
-import json
-import time
 
 from .docker_command_failed_error import DockerCommandFailedError
+from ..util.remove_whitespace import remove_whitespace
 
 
 @dataclass
@@ -21,11 +20,14 @@ def kill(self) -> None:
                 raise DockerCommandFailedError(
                     f'Kill failed with output: {stderr}')
 
-    def get_exposed_port(self, internal_port: int = 3_000) -> int:
+    def get_exposed_port(self, internal_port: int) -> int:
         docker_command = [
             'docker',
             'inspect',
-            '--format={{json .NetworkSettings.Ports}}',
+            # this is a string that cannot be break apart in a useful way
+            # pylint: disable=line-too-long
+            f'--format=\'{{{{(index (index .NetworkSettings.Ports "{internal_port}/tcp") 0).HostPort}}}}\'',
+            # pylint: enable=line-too-long
             self.id
         ]
 
@@ -39,38 +41,14 @@ def get_exposed_port(self, internal_port: int = 3_000) -> int:
             if process.returncode != 0:
                 raise DockerCommandFailedError(
                     f'Inspect failed with output: {stderr.decode("utf-8")}')
-            # Parse the JSON output to get the port mapping
-            try:
-                port_mappings = json.loads(stdout.decode('utf-8').strip())
-            except json.JSONDecodeError as e:
-                raise DockerCommandFailedError(
-                    f'Failed to decode JSON: {stdout.decode("utf-8")}. Error: {str(e)}') from e
-
-            port_key = f"{internal_port}/tcp"
-            if port_key not in port_mappings or not port_mappings[port_key]:
-                # Wait briefly and try to restart/refresh the container
-                time.sleep(1)
-                self.restart()
-                time.sleep(2)  # Wait for container to be ready
-                # Try one more time after restart
-                return self.get_exposed_port(internal_port)
 
-            try:
-                return int(port_mappings[port_key][0]['HostPort'])
-            except (KeyError, IndexError, TypeError) as e:
-                raise DockerCommandFailedError(
-                        f'Failed to parse port mapping: {stdout.decode("utf-8")}. '
-                        f'Error: {str(e)}'
-                    ) from e
+            exposed_port_as_string = remove_whitespace(
+                stdout.decode('utf-8')
+            ).replace('\'', '').replace('|', '')
 
-    def restart(self):
-        """Restart the container"""
-        docker_command = ['docker', 'restart', self.id]
-        with subprocess.Popen(
-            docker_command,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE
-        ) as process:
-            process.communicate()
-            if process.returncode != 0:
-                raise DockerCommandFailedError('Failed to restart container')
+        try:
+            return int(exposed_port_as_string)
+        except ValueError as value_error:
+            raise ValueError(
+                f'Could not parse port from: {exposed_port_as_string}'
+            ) from value_error

tests/shared/docker/container.py

@@ -1 +1 @@
-FROM ghcr.io/thenativeweb/eventsourcingdb:0.81.2
+FROM thenativeweb/eventsourcingdb:0.103.1

tests/shared/docker/eventsourcingdb/Dockerfile

@@ -17,8 +17,6 @@ def run(self, command: list[str], is_detached: bool, do_expose_ports: bool) -> C
             docker_command.append('-d')
 
         if do_expose_ports:
-            # Expose the default HTTP port (3000)
-            docker_command.extend(['-p', '3000'])
             docker_command.append('-P')
 
         docker_command.append(self.__get_full_name())
@@ -51,4 +49,4 @@ def build(self, directory: str) -> None:
                     f'Build failed with output: {stderr.decode("utf-8")}')
 
     def __get_full_name(self) -> str:
-        return f'{self.name}:{self.tag}'
+        return f'{self.name}:{self.tag}'

tests/shared/docker/image.py

@@ -58,7 +58,7 @@ async def test_supports_authorization(
 
             total_store_items_count = 4
             if observed_items_count == total_store_items_count:
-                break
+                break 
 
     @staticmethod
     @pytest.mark.asyncio