generated from coulomb/repo-seed
56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
"""EventEnvelope — schema for state-hub lifecycle events published to NATS.
|
|
|
|
Mirrors the EventEnvelope contract defined in activity-core
|
|
(`src/activity_core/models.py`). The state-hub publishes; activity-core
|
|
consumes and routes to ActivityDefinitions.
|
|
|
|
Subject naming convention (see docs/nats-event-subjects.md):
|
|
|
|
org.statehub.{noun}.{verb}
|
|
|
|
Examples:
|
|
org.statehub.repo.registered
|
|
org.statehub.workstream.completed
|
|
org.statehub.decision.resolved
|
|
org.statehub.domain.goal.activated
|
|
org.statehub.task.stale
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
PUBLISHER = "state-hub"
|
|
|
|
|
|
class EventEnvelope(BaseModel):
|
|
"""Standard envelope shared with activity-core. Do not break compatibility.
|
|
|
|
All inbound events on activity-core's side are normalised into this shape.
|
|
"""
|
|
|
|
id: str = Field(description="UUID v4 — stable unique ID for deduplication.")
|
|
type: str = Field(description="Dot-namespaced event type, e.g. 'org.statehub.repo.registered'.")
|
|
version: str = Field(default="1.0", description="Schema version string.")
|
|
timestamp: datetime = Field(description="When the event occurred (UTC).")
|
|
publisher: str = Field(default=PUBLISHER, description="Originating service.")
|
|
attributes: dict[str, Any] = Field(
|
|
default_factory=dict,
|
|
description="Event-specific attributes; structure varies by event type.",
|
|
)
|
|
|
|
@classmethod
|
|
def new(cls, event_type: str, attributes: dict[str, Any] | None = None) -> "EventEnvelope":
|
|
"""Construct an envelope with a fresh UUID and current UTC timestamp."""
|
|
return cls(
|
|
id=str(uuid.uuid4()),
|
|
type=event_type,
|
|
timestamp=datetime.now(tz=timezone.utc),
|
|
publisher=PUBLISHER,
|
|
attributes=attributes or {},
|
|
)
|