Switch from cloudevents to cloudevents-pydantic package

Signed-off-by: Federico Busetti <[email protected]>

Author: febus982

Makefile

@@ -48,6 +48,7 @@ lint:
 	poetry run ruff check .
 
 fix:
+	poetry run ruff format .
 	poetry run ruff check . --fix
 	poetry run ruff format .
 

poetry.lock

@@ -23,7 +23,7 @@ aiosqlite = ">=0.18.0"
 alembic = "^1.11.1"
 asgiref = "^3.7.2"
 celery =  { version = "^5.3.1", extras = ["redis"] }
-cloudevents = "^1.9.0"
+cloudevents-pydantic = "^0.0.2"
 dependency-injector = { version = "^4.41.0", extras = ["pydantic"] }
 httpx = ">=0.23.0"
 opentelemetry-distro = { version = "*", extras = ["otlp"] }

pyproject.toml

@@ -1,8 +1,9 @@
 from collections.abc import Iterable, Mapping
 from typing import Any, List, Literal, Protocol, Tuple, Union
 
+from cloudevents_pydantic.events import CloudEvent
+
 from domains.books._models import BookModel
-from domains.common.cloudevent_base import BaseEvent
 
 
 class BookRepositoryInterface(Protocol):
@@ -18,4 +19,4 @@ async def find(
 
 
 class BookEventGatewayInterface(Protocol):
-    async def emit(self, event: BaseEvent) -> None: ...
+    async def emit(self, event: CloudEvent) -> None: ...

src/domains/books/_gateway_interfaces.py

@@ -48,7 +48,7 @@ async def create_book(self, book: BookData) -> Book:
         book_cpu_intensive_task.delay(book_id=book.book_id)
 
         await self._event_gateway.emit(
-            BookCreatedV1(
+            BookCreatedV1.event_factory(
                 data=BookCreatedV1Data.model_validate(book_model, from_attributes=True)
             )
         )

src/domains/books/_service.py

@@ -1,13 +1,10 @@
-import typing
+from typing import Annotated, Literal
 
 import pydantic
-from pydantic import ConfigDict
-
-from domains.common.cloudevent_base import (
-    BaseEvent,
-    dataschema_field,
-    type_field,
-)
+from cloudevents_pydantic.events import CloudEvent
+from cloudevents_pydantic.events.fields import metadata
+from cloudevents_pydantic.events.fields.types import URI, URIReference
+from pydantic import ConfigDict, Field
 
 
 class BookCreatedV1Data(pydantic.BaseModel):
@@ -16,22 +13,76 @@ class BookCreatedV1Data(pydantic.BaseModel):
     author_name: str
 
 
-class BookCreatedV1(BaseEvent):
-    type: typing.Literal["book.created.v1"] = type_field("book.created.v1")
-    dataschema: str = dataschema_field("book.created.v1")
+def _dataschema_url(value: str) -> str:
+    return f"https://this_service/dataschemas/{value}"
+
+
+class BookCreatedV1(CloudEvent):
+    source: Annotated[
+        URIReference,
+        Field(default="/book_service", validate_default=True),
+        metadata.FieldSource,
+    ]
+    type: Annotated[
+        Literal["book.created.v1"], Field(default="book.created.v1"), metadata.FieldType
+    ]
+    dataschema: Annotated[
+        URI,
+        Field(default=_dataschema_url("book.created.v1"), validate_default=True),
+        metadata.FieldDataSchema,
+    ]
 
     data: BookCreatedV1Data
 
     # The first example is used to generate the OpenAPI documentation!
-    # Examples ate good! Add examples!
+    # Examples are good! Add examples!
     model_config = ConfigDict(
         json_schema_extra={
             "examples": [
                 {
                     "source": "this.service.url.here",
                     "type": "book.created.v1",
-                    "dataschema": "book.created.v1/some_event",
-                    "datacontenttype": "text/xml",
+                    "dataschema": "/dataschemas/book.created.v1",
+                    "datacontenttype": "application/json",
+                    "subject": "123",
+                    "data": {"book_id": 0, "title": "string", "author_name": "string"},
+                    "id": "A234-1234-1234",
+                    "specversion": "1.0",
+                    "time": "2018-04-05T17:31:00Z",
+                }
+            ]
+        }
+    )
+
+
+class BookUpdatedV1(CloudEvent):
+    source: Annotated[
+        URIReference,
+        Field(default="/book_service", validate_default=True),
+        metadata.FieldSource,
+    ]
+    type: Annotated[
+        Literal["book.updated.v1"], Field(default="book.updated.v1"), metadata.FieldType
+    ]
+    dataschema: Annotated[
+        URI,
+        Field(default=_dataschema_url("book.updated.v1"), validate_default=True),
+        metadata.FieldDataSchema,
+    ]
+
+    # This is just an example, too lazy to use a different data class
+    data: BookCreatedV1Data
+
+    # The first example is used to generate the OpenAPI documentation!
+    # Examples are good! Add examples!
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {
+                    "source": "this.service.url.here",
+                    "type": "book.updated.v1",
+                    "dataschema": "/dataschemas/book.updated.v1",
+                    "datacontenttype": "application/json",
                     "subject": "123",
                     "data": {"book_id": 0, "title": "string", "author_name": "string"},
                     "id": "A234-1234-1234",

src/domains/books/events.py

@@ -1,11 +1,10 @@
+from cloudevents_pydantic.events import CloudEvent
 from structlog import get_logger
 
-from domains.common.cloudevent_base import BaseEvent
-
 
 class NullEventGateway:
     async def emit(
-        self, event: BaseEvent
+        self, event: CloudEvent
     ) -> None:  # pragma: no cover # No need to test this
         logger = get_logger()
         await logger.ainfo(