cloud_event.py

import json
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Any

@dataclass
class CloudEvent:
    id: str
    type: str
    source: str
    time: str
    data: Any

    @staticmethod
    def wrap(obj: Any, *, event_type: str, source: str, id: str) -> bytes:
        evt = CloudEvent(
            id=id,
            type=event_type or (obj.__class__.__name__ if obj is not None else "NullOrEmpty"),
            source=source,
            time=datetime.now(timezone.utc).isoformat(),
            data=obj,
        )
        return json.dumps(asdict(evt), ensure_ascii=False).encode("utf-8")