spanforge.event

Core event envelope and tag container for spanforge.

This module provides the Event class (the immutable event envelope) and the Tags class (an immutable str → str mapping for arbitrary metadata).

See the Events User Guide for full usage examples.

Tags

class Tags(**kwargs: str)

An immutable, validated str → str mapping attached to an event.

All keys and values must be non-empty strings. Tags is frozen after construction — there are no mutation methods.

Args:

Raises: SchemaValidationError — if any key or value is not a non-empty string.

Example:

tags = Tags(env="production", model="gpt-4o")
tags.get("env")       # "production"
tags.to_dict()        # {"env": "production", "model": "gpt-4o"}

Methods

get(key: str, default: Optional[str] = None) -> Optional[str]

Return the value for key, or default if the key is absent.

keys() -> KeysView[str]

Return a view of all tag keys.

values() -> ValuesView[str]

Return a view of all tag values.

items() -> ItemsView[str, str]

Return a view of all (key, value) pairs.

to_dict() -> Dict[str, str]

Return a plain dict copy of the tags.

Event

class Event(
    *,
    event_type: str,
    source: str,
    payload: Dict[str, Any],
    schema_version: str = "2.0",
    event_id: Optional[str] = None,
    timestamp: Optional[str] = None,
    trace_id: Optional[str] = None,
    span_id: Optional[str] = None,
    parent_span_id: Optional[str] = None,
    org_id: Optional[str] = None,
    team_id: Optional[str] = None,
    actor_id: Optional[str] = None,
    session_id: Optional[str] = None,
    tags: Optional[Tags] = None,
    checksum: Optional[str] = None,
    signature: Optional[str] = None,
    prev_id: Optional[str] = None,
)

Immutable, validated event envelope.

All fields are read-only properties after construction. event_id and timestamp are auto-generated (ULID and UTC ISO-8601 respectively) if not provided.

Args:

Raises: SchemaValidationError — if any supplied field value is invalid.

Example:

from spanforge import Event, EventType

event = Event(
    event_type=EventType.TRACE_SPAN_COMPLETED,
    source="llm-trace@0.3.1",
    payload={"span_name": "run", "status": "ok"},
    org_id="acme",
    tags=Tags(env="production"),
)

Properties

All properties are read-only.

Methods

validate() -> None

Validate all fields against the schema rules.

Checks every field in order: schema_version, event_id, event_type, timestamp, source, payload, optional tracing IDs, optional context strings, optional integrity fields, and tags.

Raises: SchemaValidationError — on the first invalid field, with .field, .received, and .reason attributes describing the problem.

to_dict(*, omit_none: bool = True) -> Dict[str, Any]

Serialise to a plain Python dictionary.

Args:

Returns: Dict[str, Any] — dictionary representation of the event.

to_json() -> str

Serialise to a canonical, deterministic JSON string.

• Keys are sorted alphabetically at every nesting level.
• None values are omitted.
• Uses compact separators (no whitespace).
• Byte-for-byte identical for the same event on any platform.

Returns: str — compact canonical JSON string.

Raises: SerializationError — if the payload contains a non-JSON-serialisable value.

payload_checksum() -> str

Compute the SHA-256 digest of the canonical JSON payload.

Returns: str — hex-encoded SHA-256 prefixed with "sha256:".

from_dict(data: Dict[str, Any], *, max_size_bytes: int = 1_048_576, max_payload_depth: int = 10, max_tags: int = 50, source_hint: str = "<dict>") -> Event (classmethod)

Construct an Event from a plain dictionary.

The dictionary shape matches to_dict() output. The returned event is not yet validated — call validate() separately if needed.

Unrecognised keys in data are preserved in event.unknown_fields and included in to_dict() output, enabling lossless round-tripping of events that contain extension or future-version fields.

DoS limits are enforced before any field parsing (RFC §19.4). Pass 0 to disable an individual limit.

Args:

Returns: Event

Raises: DeserializationError — if a required field is missing, has an unexpected type, or any DoS limit is exceeded.

from_json(json_str: str, *, max_size_bytes: int = 1_048_576, max_payload_depth: int = 10, max_tags: int = 50, source_hint: str = "<json>") -> Event (classmethod)

Construct an Event from a JSON string (as produced by to_json()).

The returned event is not yet validated — call validate() if needed.

The byte-length of the UTF-8-encoded string is checked before parsing to guard against parse-bomb attacks (RFC §19.4). Pass max_size_bytes=0 to disable the pre-parse check.

Args:

Returns: Event

Raises: DeserializationError — if json_str is not valid JSON, is missing required fields, or any DoS limit is exceeded.