Event + Context Module¶
The event module provides domain-agnostic event and context models for cascade processing. These models work across all domains including finance, healthcare, cybersecurity, content moderation, and more.
UniversalEvent¶
Universal event structure that works across all domains.
This abstraction allows the same cascade routing logic to work for: - Financial transactions - Healthcare claims - Content moderation posts - Security access attempts - Customer support tickets - And any other domain
Attributes:
| Name | Type | Description |
|---|---|---|
id |
str
|
Unique event identifier |
domain |
DomainType
|
The domain this event belongs to |
event_type |
str
|
Type of event (transaction, post, claim, etc.) |
timestamp |
datetime
|
When the event occurred |
primary_entity |
str
|
Who initiated (user, customer, patient, employee) |
secondary_entity |
str
|
Who received/target (merchant, provider, system) |
value |
float
|
Numeric value (amount, size, count, severity 0-1) |
unit |
str
|
Unit of value (USD, bytes, items, severity_score) |
domain_data |
Dict[str, Any]
|
Domain-specific payload |
__init__(id: str, domain: DomainType, event_type: str, timestamp: datetime, primary_entity: str, secondary_entity: str, value: float, unit: str, domain_data: Dict[str, Any] = dict(), correlation_id: Optional[str] = None, source_system: Optional[str] = None, event_version: str = '1.0') -> None
¶
to_dict() -> Dict[str, Any]
¶
Convert event to dictionary.
EventContext¶
Complete contextual information surrounding an event.
This structured context enables domain-agnostic routing rules that work across finance, healthcare, security, content, etc.
to_dict() -> Dict[str, Any]
¶
Convert context to dictionary.
EventWithContext¶
Complete event package with context - the input to cascade processing.
This is the primary input type for the CascadeEngine.execute() method. It combines what happened (Event) with the circumstances (Context).
Example
event = UniversalEvent( ... id="evt_123", ... domain=DomainType.CONTENT_MODERATION, ... event_type="post", ... timestamp=datetime.now(), ... primary_entity="user_456", ... secondary_entity="forum_general", ... value=0.0, # no monetary value ... unit="post", ... domain_data={"content": "Hello world", "has_media": False} ... ) context = EventContext( ... session=SessionContext(ip_address="1.2.3.4"), ... historical=HistoricalContext(account_age_days=30, previous_events_count=100) ... ) event_with_context = EventWithContext(event=event, context=context) result = await engine.execute(event_with_context)
to_dict() -> Dict[str, Any]
¶
Convert to dictionary for cascade processing.
to_flat_dict() -> Dict[str, Any]
¶
Convert to flat dictionary for backward compatibility.
This allows the event+context to work with existing ExecutionContext that expects flat dictionary access like ctx.get("event.value").
from_dict(data: Dict[str, Any]) -> EventWithContext
classmethod
¶
Parse from dictionary (e.g., API request).