feat: add event schemas (#35)

* feat: add event schemas

* chore: satisfy linter

* fix: add missing case and update tests

Author: Noah Hummel

eventsourcingdb_client_python/client.py

@@ -7,6 +7,9 @@
 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 .handlers.read_event_types.event_type import EventType
+from .handlers.read_event_types.read_event_types import read_event_types
+from .handlers.register_event_schema.register_event_schema import register_event_schema
 from .http_client.http_client import HttpClient
 from .handlers.ping import ping
 from .handlers.read_events import read_events, ReadEventsOptions
@@ -72,6 +75,13 @@ async def read_events(
         async for event in read_events(self, subject, options):
             yield event
 
+    async def read_event_types(self) -> AsyncGenerator[EventType, None]:
+        async for event_type in read_event_types(self):
+            yield event_type
+
+    async def register_event_schema(self, event_type: str, json_schema: str) -> None:
+        await register_event_schema(self, event_type, json_schema)
+
     async def observe_events(
         self,
         subject: str,

eventsourcingdb_client_python/handlers/read_event_types/__init__.py

@@ -0,0 +1,42 @@
+from dataclasses import dataclass
+from typing import TypeVar
+
+from eventsourcingdb_client_python.errors.validation_error import ValidationError
+
+Self = TypeVar("Self", bound="EventType")
+
+
+@dataclass
+class EventType:
+    event_type: str
+    is_phantom: bool
+    schema: str | None = None
+
+    @staticmethod
+    def parse(unknown_object: dict) -> Self:
+        event_type = unknown_object.get('eventType')
+        if not isinstance(event_type, str):
+            raise ValidationError(
+                f"Failed to parse eventType '{event_type}' to str."
+            )
+
+        is_phantom = unknown_object.get('isPhantom')
+        if not isinstance(is_phantom, bool):
+            raise ValidationError(
+                f"Failed to parse isPhantom '{is_phantom}' to bool."
+            )
+
+        schema = unknown_object.get('schema')
+        if schema is not None and not isinstance(schema, str):
+            raise ValidationError(
+                f"Failed to parse schema '{schema}' to str."
+            )
+
+        return EventType(
+            event_type=event_type,
+            is_phantom=is_phantom,
+            schema=schema,
+        )
+
+    def __hash__(self):
+        return hash((self.event_type, self.is_phantom, self.schema))

eventsourcingdb_client_python/handlers/read_event_types/event_type.py

@@ -0,0 +1,12 @@
+from typing import Any
+
+EVENT_TYPE_TYPE = 'eventType'
+
+
+def is_event_type(message: Any) -> bool:
+    if not isinstance(message, dict) or message.get('type') != EVENT_TYPE_TYPE:
+        return False
+
+    payload = message.get('payload')
+
+    return isinstance(payload, dict)

eventsourcingdb_client_python/handlers/read_event_types/is_event_type.py

@@ -0,0 +1,56 @@
+from collections.abc import AsyncGenerator
+from http import HTTPStatus
+
+from .event_type import EventType
+from .is_event_type import is_event_type
+from ..is_stream_error import is_stream_error
+from ..parse_raw_message import parse_raw_message
+from ...abstract_base_client import AbstractBaseClient
+from ...errors.custom_error import CustomError
+from ...errors.internal_error import InternalError
+from ...errors.server_error import ServerError
+from ...errors.validation_error import ValidationError
+from ...http_client.response import Response
+
+
+async def read_event_types(
+    client: AbstractBaseClient,
+) -> AsyncGenerator[EventType, None]:
+    response: Response
+    try:
+        response = await client.http_client.post(
+            path='/api/read-event-types',
+            request_body='',
+        )
+    except CustomError as custom_error:
+        raise custom_error
+    except Exception as other_error:
+        raise InternalError(str(other_error)) from other_error
+
+    with response:
+        if response.status_code != HTTPStatus.OK:
+            raise ServerError(
+                'Unexpected response status: '
+                f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
+            )
+        async for raw_message in response.body:
+            message = parse_raw_message(raw_message)
+
+            if is_stream_error(message):
+                raise ServerError(message['payload']['error'])
+
+            if is_event_type(message):
+                event_type: EventType
+                try:
+                    event_type = EventType.parse(message['payload'])
+                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
+
+                yield event_type
+                continue
+
+            raise ServerError(
+                f'Failed to read event types, an unexpected stream item was received \'{message}\'.'
+            )

eventsourcingdb_client_python/handlers/read_event_types/read_event_types.py

@@ -1,4 +1,4 @@
-from collections.abc import Generator
+from collections.abc import AsyncGenerator
 import json
 from http import HTTPStatus
 
@@ -18,7 +18,7 @@
 async def read_subjects(
     client: AbstractBaseClient,
     base_subject: str
-) -> Generator[str, None, None]:
+) -> AsyncGenerator[str, None, None]:
     try:
         validate_subject(base_subject)
     except ValidationError as validation_error:
@@ -44,7 +44,7 @@ async def read_subjects(
     with response:
         if response.status_code != HTTPStatus.OK:
             raise ServerError(
-                f'Unexpected response status: '
+                'Unexpected response status: '
                 f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
             )
         async for raw_message in response.body:

eventsourcingdb_client_python/handlers/read_subjects/read_subjects.py

@@ -0,0 +1,51 @@
+import json
+from http import HTTPStatus
+
+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 ...event.validate_type import validate_type
+from ...http_client.response import Response
+
+
+async def register_event_schema(
+    client: AbstractBaseClient,
+    event_type: str,
+    json_schema: str,
+) -> None:
+    try:
+        validate_type(event_type)
+    except ValidationError as validation_error:
+        raise InvalidParameterError(
+            'event_type', str(validation_error)
+        ) from validation_error
+    except Exception as other_error:
+        raise InternalError(str(other_error)) from other_error
+
+    request_body = json.dumps({
+        'eventType': event_type,
+        'schema': json_schema,
+    })
+
+    response: Response
+    try:
+        response = await client.http_client.post(
+            path='/api/register-event-schema',
+            request_body=request_body,
+        )
+    except CustomError as custom_error:
+        raise custom_error
+    except Exception as other_error:
+        raise InternalError(str(other_error)) from other_error
+
+    with response:
+        if response.status_code != HTTPStatus.OK:
+            raise ServerError(
+                'Unexpected response status: '
+                f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
+            )
+
+    return

eventsourcingdb_client_python/handlers/register_event_schema/__init__.py

@@ -87,21 +87,36 @@ def __validate_protocol_version(self, http_status_code: int, headers: Headers) -
             f' client \'{self.__client_configuration.protocol_version}\'.'
         )
 
-    def __validate_response(
+    @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 += '.'
+
+        return error_message
+
+    async def __validate_response(
         self,
         response: Response
     ) -> RetryResult[Response]:
         server_failure_range = Range(500, 600)
         if server_failure_range.lower <= response.status_code < server_failure_range.upper:
-            return Retry(
-                ServerError(f'Request failed with status code \'{response.status_code}\'.')
-            )
+            return Retry(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(
-                f'Request failed with status code \'{response.status_code}\'.'
-            )
+            raise ClientError(await self.__get_error_message(response))
 
         self.__validate_protocol_version(response.status_code, response.headers)
 
@@ -132,7 +147,7 @@ async def execute_request() -> RetryResult[Response]:
 
             response = Response(response)
             try:
-                result = self.__validate_response(response)
+                result = await self.__validate_response(response)
             except Exception as error:
                 response.close()
                 raise error
@@ -171,7 +186,7 @@ async def execute_request() -> RetryResult[Response]:
 
             response = Response(response)
             try:
-                result = self.__validate_response(response)
+                result = await self.__validate_response(response)
             except Exception as error:
                 response.close()
                 raise error