Merge pull request #66 from febus982/fastapi-0.103

Fastapi 0.103

Author: febus982

http_app/routes/events.py

@@ -1,10 +1,12 @@
 import typing
-from typing import Any, Dict, List
+from typing import Dict, List
 
 from fastapi import APIRouter, Body, Header, HTTPException
+from fastapi.openapi.models import Example
 from pydantic import BaseModel
 
 from domains.books.events import BookCreatedV1
+from domains.books.service import BookService
 
 router = APIRouter(prefix="/events")
 
@@ -22,7 +24,7 @@ def _event_registry() -> Dict[str, typing.Type[BaseModel]]:
     }
 
 
-def _event_schema_examples() -> dict[str, dict[str, Any]]:
+def _event_schema_examples() -> dict[str, Example]:
     missing_example_message = (
         "No example has been added to this event but you can"
         " still explore the event schema. (Ask the developer"
@@ -31,11 +33,11 @@ def _event_schema_examples() -> dict[str, dict[str, Any]]:
     )
 
     return {
-        k: {
-            "value": getattr(v, "model_config", {})
+        k: Example(
+            value=getattr(v, "model_config", {})
             .get("json_schema_extra", {})
             .get("examples", [missing_example_message])[0]
-        }
+        )
         for k, v in _event_registry().items()
     }
 
@@ -67,15 +69,6 @@ async def event_schema_list() -> List[str]:
 
 @router.post(
     "",
-    openapi_extra={
-        "requestBody": {
-            "content": {
-                "application/cloudevents+json; charset=UTF-8": {
-                    "examples": _event_schema_examples(),
-                }
-            },
-        },
-    },
     status_code=204,
     description="""
     Entrypoint for CloudEvent processing, it supports only single events.
@@ -86,10 +79,12 @@ async def event_schema_list() -> List[str]:
 async def submit_event(
     event_data: _EVENTS_UNION_TYPE = Body(
         media_type="application/cloudevents+json; charset=UTF-8",
+        openapi_examples=_event_schema_examples(),
         discriminator="type",
     ),
     content_type: typing.Literal[
         "application/cloudevents+json; charset=UTF-8"
     ] = Header(),
 ) -> None:
-    pass
+    # Some routing will be necessary when multiple event types will be supported
+    await BookService().book_created_event_handler(event_data.data.book_id)

poetry.lock

@@ -1,7 +1,8 @@
-from unittest.mock import patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 from httpx import AsyncClient
 
+from domains.books import service
 from domains.books.events import BookCreatedV1
 from domains.common.cloudevent_base import BaseEvent
 
@@ -39,12 +40,16 @@ async def test_event_returns_204(testapp):
     fake_event = BookCreatedV1(
         data={"book_id": 0, "title": "string", "author_name": "string"},
     )
-    async with AsyncClient(app=testapp, base_url="http://test") as ac:
-        response = await ac.post(
-            "/events",
-            headers={"content-type": "application/cloudevents+json; charset=UTF-8"},
-            content=fake_event.model_dump_json(),
-        )
+    svc = MagicMock(autospec=service.BookService)
+    svc.book_created_event_handler = AsyncMock(return_value=None)
+    with patch("domains.books.service.BookService.__new__", return_value=svc):
+        async with AsyncClient(app=testapp, base_url="http://test") as ac:
+            response = await ac.post(
+                "/events",
+                headers={"content-type": "application/cloudevents+json; charset=UTF-8"},
+                content=fake_event.model_dump_json(),
+            )
+    svc.book_created_event_handler.assert_called_once()
     assert response.status_code == 204