From c8e6f9dcf9423289655ed9a469a521cdb840ef90 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Thu, 16 Jul 2026 12:37:29 +0300 Subject: [PATCH] pubsub: Remove cloudevents dependency The pinned cloudevents 2.0.0 no longer provides the cloudevents.http module, so api/pubsub.py and api/pubsub_mongo.py fail to import and the API cannot start. Instead of migrating to the new cloudevents v2 API, drop the library: it was only used to fill in the "id", "time" and "specversion" attributes and serialize the event to JSON. Add a small to_cloudevent_json() helper that emits the exact same CloudEvents 1.0 JSON envelope (verified byte-identical against cloudevents 1.12.1 output for dict data, string data and extension attributes such as "owner"), so the wire format consumed by external subscribers is unchanged. Signed-off-by: Denys Fedoryshchenko --- api/pubsub.py | 33 +++++++++++++++++++++----- api/pubsub_mongo.py | 16 ++++++------- docker/api/requirements.txt | 1 - pyproject.toml | 1 - tests/e2e_tests/test_pipeline.py | 7 +++--- tests/e2e_tests/test_pubsub_handler.py | 17 +++++++------ 6 files changed, 46 insertions(+), 29 deletions(-) diff --git a/api/pubsub.py b/api/pubsub.py index 3535450b..92adeca6 100644 --- a/api/pubsub.py +++ b/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""" diff --git a/api/pubsub_mongo.py b/api/pubsub_mongo.py index ff7f3b8b..ffe2f299 100644 --- a/api/pubsub_mongo.py +++ b/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""" diff --git a/docker/api/requirements.txt b/docker/api/requirements.txt index d2a81106..48ddf6b6 100644 --- a/docker/api/requirements.txt +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index e16d8d0b..ac6c5c9a 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/e2e_tests/test_pipeline.py b/tests/e2e_tests/test_pipeline.py index 55262539..96ef6106 100644 --- a/tests/e2e_tests/test_pipeline.py +++ b/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", diff --git a/tests/e2e_tests/test_pubsub_handler.py b/tests/e2e_tests/test_pubsub_handler.py index 3015b7ac..5a55777d 100644 --- a/tests/e2e_tests/test_pubsub_handler.py +++ b/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"