pubsub: Remove cloudevents dependency

api/pubsub.py

@@ -8,9 +8,9 @@
 import asyncio
 import json
 import logging
-from datetime import datetime, timedelta
+import uuid
+from datetime import datetime, timedelta, timezone
 
-from cloudevents.http import CloudEvent, to_json
 from redis import asyncio as aioredis
 
 from .config import PubSubSettings
@@ -19,6 +19,29 @@
 logger = logging.getLogger(__name__)
 
 
+def to_cloudevent_json(data, attributes) -> bytes:
+    """Serialize an event as a CloudEvents 1.0 JSON envelope
+
+    Produce the same JSON as the cloudevents library used to: the
+    "specversion", "id" and "time" attributes are auto-generated if
+    not provided and extension attributes (e.g. "owner") are emitted
+    at the top level after "data".
+    """
+    event = {
+        "specversion": attributes.get("specversion", "1.0"),
+        "id": attributes.get("id") or str(uuid.uuid4()),
+        "source": attributes.get("source"),
+        "type": attributes.get("type"),
+        "time": attributes.get("time")
+        or datetime.now(timezone.utc).isoformat(),
+        "data": data,
+    }
+    for key, value in attributes.items():
+        if key not in event:
+            event[key] = value
+    return json.dumps(event).encode("utf-8")
+
+
 class PubSub:
     """Pub/Sub implementation class
 
@@ -214,8 +237,7 @@ async def publish_cloudevent(self, channel, data, attributes=None):
             attributes["type"] = "api.kernelci.org"
         if not attributes.get("source"):
             attributes["source"] = self._settings.cloud_events_source
-        event = CloudEvent(attributes=attributes, data=data)
-        await self.publish(channel, to_json(event))
+        await self.publish(channel, to_cloudevent_json(data, attributes))
 
     async def push_cloudevent(self, list_name, data, attributes=None):
         """Push a CloudEvent on a list
@@ -230,8 +252,7 @@ async def push_cloudevent(self, list_name, data, attributes=None):
                 "type": "api.kernelci.org",
                 "source": self._settings.cloud_events_source,
             }
-        event = CloudEvent(attributes=attributes, data=data)
-        await self.push(list_name, to_json(event))
+        await self.push(list_name, to_cloudevent_json(data, attributes))
 
     async def subscription_stats(self):
         """Get existing subscription details"""

api/pubsub_mongo.py

@@ -23,12 +23,12 @@
 from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional
 
-from cloudevents.http import CloudEvent, to_json
 from pymongo import ASCENDING, AsyncMongoClient, WriteConcern
 from redis import asyncio as aioredis
 
 from .config import PubSubSettings
 from .models import SubscriberState, Subscription, SubscriptionStats
+from .pubsub import to_cloudevent_json
 
 logger = logging.getLogger(__name__)
 
@@ -235,8 +235,7 @@ async def _publish_keepalive(self, channel: str, data: str):
             "type": "api.kernelci.org",
             "source": self._settings.cloud_events_source,
         }
-        event = CloudEvent(attributes=attributes, data=data)
-        await self._redis.publish(channel, to_json(event))
+        await self._redis.publish(channel, to_cloudevent_json(data, attributes))
 
     def _update_channels(self):
         """Update tracked channels from active subscriptions"""
@@ -323,8 +322,9 @@ def _eventhistory_to_cloudevent(self, event: Dict) -> str:
         if event.get("owner"):
             attributes["owner"] = event["owner"]
 
-        ce = CloudEvent(attributes=attributes, data=event.get("data", {}))
-        return to_json(ce).decode("utf-8")
+        return to_cloudevent_json(event.get("data", {}), attributes).decode(
+            "utf-8"
+        )
 
     async def _get_missed_events(
         self,
@@ -645,8 +645,7 @@ async def publish_cloudevent(
         sequence_id = await self._store_event(channel, data, owner)
 
         # Create CloudEvent for Redis real-time delivery
-        event = CloudEvent(attributes=attributes, data=data)
-        event_json = to_json(event).decode("utf-8")
+        event_json = to_cloudevent_json(data, attributes).decode("utf-8")
 
         # Add sequence_id to message for tracking durable subscriptions
         msg_with_id = json.loads(event_json)
@@ -674,8 +673,7 @@ async def push_cloudevent(
                 "type": "api.kernelci.org",
                 "source": self._settings.cloud_events_source,
             }
-        event = CloudEvent(attributes=attributes, data=data)
-        await self.push(list_name, to_json(event))
+        await self.push(list_name, to_cloudevent_json(data, attributes))
 
     async def subscription_stats(self) -> List[SubscriptionStats]:
         """Get existing subscription details"""

docker/api/requirements.txt

@@ -1,4 +1,3 @@
-cloudevents==2.0.0
 beanie==2.1.0
 fastapi[all]==0.135.3
 fastapi-pagination==0.14.3

pyproject.toml

@@ -16,7 +16,6 @@ readme = "README.md"
 requires-python = ">=3.10"
 license = {text = "LGPL-2.1-or-later"}
 dependencies = [
-  "cloudevents == 2.1.0",
   "beanie == 2.1.0",
   "fastapi[all] == 0.136.1",
   "fastapi-pagination == 0.15.12",

tests/e2e_tests/test_pipeline.py

@@ -6,8 +6,9 @@
 
 """Run test pipeline for KernelCI API"""
 
+import json
+
 import pytest
-from cloudevents.http import from_json
 
 from .listen_handler import create_listen_task
 from .test_node_handler import (
@@ -68,7 +69,7 @@ async def test_node_pipeline(test_async_client):
 
     # Get result of pubsub event listen task
     await task_listen
-    event_data = from_json(task_listen.result().json().get("data")).data
+    event_data = json.loads(task_listen.result().json().get("data"))["data"]
     assert event_data != "BEEP"
     keys = {
         "op",
@@ -103,7 +104,7 @@ async def test_node_pipeline(test_async_client):
 
     # Get result of pubsub event listen task
     await task_listen
-    event_data = from_json(task_listen.result().json().get("data")).data
+    event_data = json.loads(task_listen.result().json().get("data"))["data"]
     assert event_data != "BEEP"
     keys = {
         "op",

tests/e2e_tests/test_pubsub_handler.py

@@ -6,8 +6,9 @@
 
 """End-to-end test function for KernelCI API pubsub handler"""
 
+import json
+
 import pytest
-from cloudevents.http import CloudEvent, from_json, to_structured
 
 from .listen_handler import create_listen_task
 
@@ -30,17 +31,15 @@ async def test_pubsub_handler(test_async_client):
         test_async_client, pytest.test_channel_subscription_id
     )
 
-    # Created and publish CloudEvent
-    attributes = {
+    # Create and publish an event
+    event = {
         "type": "api.kernelci.org",
         "source": "https://api.kernelci.org/",
+        "data": {"message": "Test message"},
     }
-    data = {"message": "Test message"}
-    event = CloudEvent(attributes, data)
-    headers, body = to_structured(event)
-    headers["Authorization"] = f"Bearer {pytest.BEARER_TOKEN}"
+    headers = {"Authorization": f"Bearer {pytest.BEARER_TOKEN}"}
     response = await test_async_client.post(
-        "publish/test_channel", headers=headers, data=body
+        "publish/test_channel", headers=headers, json=event
     )
     assert response.status_code == 200
 
@@ -52,7 +51,7 @@ async def test_pubsub_handler(test_async_client):
         "pattern",
         "type",
     }
-    event_data = from_json(task_listen.result().json().get("data")).data
+    event_data = json.loads(task_listen.result().json().get("data"))["data"]
     assert event_data != "BEEP"
     assert ("message",) == tuple(event_data.keys())
     assert event_data.get("message") == "Test message"