feat: Internal API now has the concept of a 'routing_key'

Author: Alan Christie

.pylintrc

@@ -0,0 +1,5 @@
+[messages control]
+
+# Disable some distracting checks.
+# See http://pylint.pycqa.org/en/latest/user_guide/message-control.html
+disable = too-few-public-methods,

README.md

@@ -41,6 +41,20 @@ current health of your clone with: -
 
     pre-commit run --all-files
 
+## Local execution
+You can build and run the service using `docker compose`: -
+
+    docker compose up --build --detach
+
+And then interact with it using `http`, where you should be able to create
+and delete event streams using the internal API. Here we're using
+`jq` to process the response body to simplify the subsequent **DELETE** request: -
+
+    ES_ID=$(http post localhost:8081/event-stream/ routing_key=0123456789 -b | jq -r ".id")
+    echo $ES_ID
+
+    http delete localhost:8081/event-stream/$ES_ID -b
+
 ---
 
 [black]: https://black.readthedocs.io/en/stable

app/app.py

@@ -1,14 +1,25 @@
 """The entrypoint for the Squonk2 FastAPI WebSocket service."""
 
 import uuid
+from typing import Dict
 
 from fastapi import FastAPI, HTTPException, WebSocket, status
+from pydantic import BaseModel
 
 # Public (event-stream) and internal (REST) services
 app_public = FastAPI()
 app_internal = FastAPI()
 
-_ES_UUID_SET: set = set()
+# A map of event stream IDs (a UUID with a "es-" prefix) to their routing keys.
+_ES_UUID_MAP: Dict = {}
+
+
+# We use pydantic to declare the model (request payloads) for the internal REST API.
+# The public API is a WebSocket API and does not require a model.
+class EventStreamPostRequestBody(BaseModel):
+    """/event-stream/ POST request body."""
+
+    routing_key: str
 
 
 # Endpoints for the 'public-facing' event-stream web-socket API ------------------------
@@ -17,9 +28,9 @@
 @app_public.websocket("/event-stream/{es_id}")
 async def event_stream(websocket: WebSocket, es_id: str):
     """The websocket handler for the event-stream."""
-    if not es_id in _ES_UUID_SET:
+    if not es_id in _ES_UUID_MAP:
         raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
+            status_code=status.HTTP_404_NOT_FOUND,
             detail=f"EventStream {es_id} is not known",
         )
     await websocket.accept()
@@ -32,20 +43,24 @@ async def event_stream(websocket: WebSocket, es_id: str):
 
 
 @app_internal.post("/event-stream/")
-def post_es():
+def post_es(request_body: EventStreamPostRequestBody):
     """Create a new event-stream."""
     uuid_str = f"es-{uuid.uuid4()}"
-    _ES_UUID_SET.add(uuid_str)
+    _ES_UUID_MAP[uuid_str] = request_body.routing_key
+    print(_ES_UUID_MAP)
+
     return {"id": uuid_str}
 
 
 @app_internal.delete("/event-stream/{es_id}")
 def delete_es(es_id: str):
     """Destroys an existing event-stream."""
-    if es_id not in _ES_UUID_SET:
+    print(_ES_UUID_MAP)
+    if es_id not in _ES_UUID_MAP:
         raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
+            status_code=status.HTTP_404_NOT_FOUND,
             detail=f"EventStream {es_id} is not known",
         )
-    _ES_UUID_SET.remove(es_id)
+    _ = _ES_UUID_MAP.pop(es_id)
+
     return {}

poetry.lock

@@ -2,15 +2,16 @@
 name = "squonk2-fastapi-ws-event-stream"
 version = "1.0.0"
 package-mode = false
-description = "A Python FastAPI Squonk2 AS EvenStream Service"
+description = "A Squonk2 Python FastAPI AS EvenStream Service"
 authors = ["Alan Christie <[email protected]>"]
 readme = "README.md"
 
 [tool.poetry.dependencies]
 # Locked versions (for now)
 fastapi = "0.115.8"
 uvicorn = "0.34.0"
-
+# Flexible versions
+pydantic = "^2.10.6"
 python = "^3.12"
 
 [tool.poetry.group.dev.dependencies]