Bases: BaseEvent
Structured notification from an observer.
Replaces the old Signal event. Same fields, clearer name. Produced by observers, consumed by AlertPolicy.
Source code in autogen/beta/events/base.py
| def __init__(self, *args: Any, **kwargs: Any) -> None:
# MRO walk: map positional args and collect defaults.
positional_names: list[str] = []
defaults: dict[str, Any] = {}
seen: set[str] = set()
for klass in reversed(type(self).__mro__):
for name, f in getattr(klass, "_event_fields_", {}).items():
if name not in seen:
if not f.kw_only:
positional_names.append(name)
if name not in kwargs:
default = f.get_default()
if default is not Ellipsis:
defaults[name] = default
seen.add(name)
if args:
if len(args) > len(positional_names):
raise TypeError(
f"{type(self).__name__}() takes {len(positional_names)} "
f"positional argument(s) but {len(args)} were given"
)
for name, value in zip(positional_names, args):
if name in kwargs:
raise TypeError(f"{type(self).__name__}() got multiple values for argument '{name}'")
kwargs[name] = value
defaults.pop(name, None)
# Apply defaults first, then user-provided kwargs so that
# property setters (e.g. content -> _content) aren't overwritten
# by a field default applied afterwards.
for key, value in defaults.items():
setattr(self, key, value)
for key, value in kwargs.items():
setattr(self, key, value)
|
instance-attribute
instance-attribute
instance-attribute
class-attribute instance-attribute
class-attribute instance-attribute
created_at = Field(default_factory=time, compare=False, repr=False)
Serialize this event to a JSON-compatible dictionary.
Source code in autogen/beta/events/base.py
| def to_dict(self) -> dict[str, Any]:
"""Serialize this event to a JSON-compatible dictionary."""
return event_to_dict(self)
|
classmethod
Reconstruct an event from a serialized dictionary.
Filters input to only fields known by this class (via MRO Field descriptors), then constructs via cls(**filtered).
Source code in autogen/beta/events/base.py
| @classmethod
def from_dict(cls, data: dict[str, Any]) -> "BaseEvent":
"""Reconstruct an event from a serialized dictionary.
Filters input to only fields known by this class (via MRO Field
descriptors), then constructs via ``cls(**filtered)``.
"""
# Collect known field names across the MRO
known_fields: set[str] = set()
for klass in cls.__mro__:
for name, f in getattr(klass, "_event_fields_", {}).items():
known_fields.add(name)
# Deserialize nested events/special types, then filter to known fields
deserialized = deserialize_payload(data)
filtered = {k: v for k, v in deserialized.items() if k in known_fields}
return cls(**filtered)
|