from dataclasses import _process_class, dataclass, field, replace, fields, MISSING, Field # type: ignore[attr-defined]
from datetime import datetime
from enum import IntFlag, auto
from http import HTTPMethod
from typing import Optional, Any, Type, TypeVar, Callable, overload, dataclass_transform
from uuid import UUID, uuid4
from zlib import adler32
from edri.api.dataclass import File, Header, Cookie, Scope
from edri.config.constant import ApiType
from edri.dataclass.response import Response, ResponseStatus
from edri.utility.function import snake2camel
@dataclass
class ApiInfo:
_key: str
type: ApiType = field(compare=False)
@property
def key(self) -> str:
return self._key
[docs]
@dataclass
class SwitchInfo:
"""Per-hop transport state for distributing an event across the Switch.
``router_id`` records which router sent the event on the current hop;
``received`` flags an event that arrived from the Switch so the local
Router does not echo it straight back (see :meth:`Router.send`). Both are
re-stamped on every send/receive and reset when the event turns into a
response.
The event's own stable identity (used by :meth:`Cache.last_events`/
:meth:`Cache.events_from` as the Switch resume marker) is
:attr:`Event._id` — every event carries one unconditionally, not just
shareable ones, so it no longer needs to live here.
"""
router_id: Optional[UUID] = None
received: bool = False
@dataclass
class TimingEvent:
name: str
description: str | None
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class Timing:
events: list[TimingEvent] = field(default_factory=list)
def stamp(self, name: str, description: str | None = None) -> None:
self.events.append(TimingEvent(name=name, description=description))
[docs]
@dataclass
class Event:
_api: Optional[ApiInfo] = field(init=False, default=None)
_switch: Optional[SwitchInfo] = field(init=False, default=None, compare=False)
_worker: str | None = field(init=False, default=None)
_id: UUID = field(init=False, compare=False, default_factory=uuid4)
_timing: Timing = field(init=False, default_factory=Timing, compare=False)
_is_file_only: str | None = field(default=None, init=False)
response: Optional[Response] = field(init=False, default=None)
method: Optional[HTTPMethod] = field(init=False, default=None)
def __post_init__(self) -> None:
# Shareable events cross the Switch and are cached on every router for
# replay after a brief outage. Stamp each instance with a SwitchInfo at
# construction so per-hop transport state exists from the moment it does
# — it must exist before the event is first cached or forked to the
# Connector process, hence at construction. The resume marker itself is
# `_id` (every event, not just shareable ones).
if type(self) in shareable_events:
self._switch = SwitchInfo()
[docs]
def get_response(self) -> Optional[Response]:
if self.response and self.response.get_status() != ResponseStatus.NONE:
return self.response
return None
[docs]
def has_response(self) -> bool:
return self.response is not None and self.response.get_status() != ResponseStatus.NONE
def _reset_switch_transport(self) -> None:
# Clear per-hop transport state so the event is re-stamped on its way
# back out (the event's stable identity is `_id`, untouched here). See
# SwitchInfo. Non-shareable events never carry a SwitchInfo (stays None).
if self._switch is not None:
self._switch.router_id = None
self._switch.received = False
[docs]
def set_response(self, response: Response) -> None:
self._reset_switch_transport()
self.response = response
[docs]
def remove_response(self) -> None:
self._reset_switch_transport()
self.response = None
[docs]
def get_context(self) -> dict[str, Any]:
"""
Retrieve a dictionary of all keyword-only fields from the dataclass.
This method inspects the dataclass and extracts all fields that are
marked as `kw_only=True`, returning their names and current values.
Returns:
dict[str, Any]: A dictionary where the keys are the names of keyword-only
fields and the values are their corresponding values
in the dataclass instance.
"""
return {
f.name: getattr(self, f.name) for f in fields(self)
if f.kw_only # Check if the field is keyword-only
}
[docs]
@classmethod
def hash_name(cls) -> int:
return adler32(f"{cls.__module__}.{cls.__qualname__}".encode())
[docs]
def copy_sender(self, event: "Event") -> None:
if event._api:
self._api = replace(event._api)
[docs]
def as_dict(self, /, *, transform: bool, keep_concealed: bool) -> dict[str, Any]:
response = self.response
# Shallow field extraction, not dataclasses.asdict(): asdict() recurses
# and copy.deepcopy()s non-primitive values, which is real, avoidable
# cost for a dict that's about to be JSON-serialized and discarded —
# matches the pattern Response.as_dict() already uses.
event = {f.name: getattr(self, f.name) for f in fields(self)}
event.pop("method", None)
event.pop("response", None)
if transform:
event = {snake2camel(key, exceptions=("_", "id_")): value for key, value in event.items() if
keep_concealed or not key.startswith("_")}
else:
event = {key: value for key, value in event.items() if keep_concealed or not key.startswith("_")}
if response:
event["response"] = response.as_dict(transform=transform, keep_concealed=keep_concealed)
return event
[docs]
class EventHandlingType(IntFlag):
"""
IntFlag representing different modes of event handling in the system.
The `EventHandlingType` IntFlag categorizes the ways in which events are handled
and dispatched to clients in the API broker system. Each mode defines a distinct
strategy for event delivery, ensuring that events reach their intended recipients
based on the specified criteria. With IntFlag, multiple modes can be combined
using bitwise operations.
Attributes:
SPECIFIC (int): Handles an event in a 1:1 fashion, responding directly to the
client that requested it. This mode ensures that the event is sent only to
the client whose request triggered the event. Events in this mode are created
exclusively by API handlers to fulfill specific client requests.
Example APIs: Typically used for REST or HTML requests where a client
makes a direct request, and the server provides a direct response.
SUBSCRIBED (int): Sends the event exclusively to clients who have registered
their interest by subscribing to this event type. This mode is used for
events meant to notify or update only a specific subset of clients.
Example APIs: Suitable for APIs with asynchronous capabilities, such as
WebSocket or MQTT, where clients subscribe to events and receive
notifications asynchronously.
ALL (int): Broadcasts the event to all clients connected to the system capable
of receiving asynchronous events, regardless of their specific subscription status.
This mode does not limit the broadcast to only those who subscribed to a specific type.
Example APIs: Similar to `SUBSCRIBED`, this mode applies to protocols like
WebSocket or MQTT, but the event is sent to all clients, not just those
subscribed to a specific type.
Combinations:
SPECIFIC|SUBSCRIBED (int): A flexible mode of event handling where the event can either
be a direct 1:1 response to a client's request or be broadcasted to all clients
subscribed to that event type. The handling behavior is determined by the context
in which the event is created. If the event originates from a REST request or similar,
it might behave like a SPECIFIC event. If generated internally as part of core logic,
it might be broadcast to subscribers.
Example APIs: Can be used for REST or HTML requests but also applies to
asynchronous protocols like WebSocket or MQTT where clients can subscribe
to specific events.
SPECIFIC|ALL (int): Similar to `SPECIFIC|SUBSCRIBED`, but with a broader scope.
The event can either be sent as a direct response to the specific client that
requested it or broadcasted to all clients capable of receiving events,
regardless of their subscription status. The handling behavior is determined
by the event's context — it is either specific or for all, but it cannot be both
at the same time.
Example APIs: Applies to REST, HTML, WebSocket, or MQTT. The event could either
act like a direct response to a client request or be broadcast to every client,
but not both simultaneously.
"""
SPECIFIC = auto()
SUBSCRIBED = auto()
ALL = auto()
shareable_events: set[Type[Event]] = set()
ignored_defaults = (Header, Cookie, Scope)
ignored_fields = ("response", "method")
_T = TypeVar("_T", bound=Event)
def _event(cls: type[_T], init: bool, repr: bool, eq: bool, order: bool, unsafe_hash: bool, frozen: bool,
match_args: bool, kw_only: bool, slots: bool, weakref_slot: bool | None, shareable: bool) -> type[_T]:
event_response = cls.__annotations__.get("response", False)
if event_response:
if not issubclass(event_response, Response):
raise AttributeError("Event response must be a subclass of Response")
attribute = getattr(cls, "response", None)
if isinstance(attribute, Field):
attribute.init = False
attribute.default_factory = event_response
else:
setattr(cls, "response", field(init=False, default_factory=event_response))
dataclass: type[_T] = _process_class(cls, init=init, repr=repr, eq=eq, order=order, unsafe_hash=unsafe_hash, frozen=frozen,
match_args=match_args, kw_only=kw_only, slots=slots, weakref_slot=weakref_slot)
valid_fields = []
for f in fields(dataclass):
if f.name.startswith('_') or f.name in ignored_fields:
continue
if f.kw_only:
continue
# Ignore fields with default or default_factory of Header, Cookie, Scope
default = f.default
default_factory = f.default_factory if f.default_factory is not MISSING else None
if isinstance(default, ignored_defaults) or isinstance(default_factory, ignored_defaults):
continue
valid_fields.append(f)
if len(valid_fields) == 1:
sole_field = valid_fields[0]
if cls.__annotations__.get(sole_field.name) == File:
dataclass._is_file_only = sole_field.name
if shareable:
if dataclass._is_file_only:
raise ValueError(
f"Event '{cls.__name__}' cannot be both file-only (upload) and shareable: "
"a file-only event is streamed through a host-local SharedMemoryPipe whose "
"reader cannot be resolved across the Switch on a remote node."
)
shareable_events.add(dataclass)
return dataclass
@dataclass_transform(field_specifiers=(field,))
@overload
def event(cls: type[_T], /) -> type[_T]: ...
@dataclass_transform(field_specifiers=(field,))
@overload
def event(cls: None = ..., /, *, init: bool = ..., repr: bool = ..., eq: bool = ..., order: bool = ...,
unsafe_hash: bool = ..., frozen: bool = ..., match_args: bool = ..., kw_only: bool = ...,
slots: bool = ..., weakref_slot: bool = ..., shareable: bool = ...) -> Callable[[type[_T]], type[_T]]: ...
[docs]
@dataclass_transform(field_specifiers=(field,))
def event(cls: type[_T] | None = None, /, *, init: bool = True, repr: bool = True, eq: bool = True, order: bool = False,
unsafe_hash: bool = False, frozen: bool = False, match_args: bool = True, kw_only: bool = False,
slots: bool = False, weakref_slot: bool = False, shareable: bool = False) -> type[_T] | Callable[[type[_T]], type[_T]]:
def wrapper(cls: type[_T]) -> type[_T]:
return _event(cls, init=init, repr=repr, eq=eq, order=order, unsafe_hash=unsafe_hash, frozen=frozen, match_args=match_args,
kw_only=kw_only, slots=slots, weakref_slot=weakref_slot, shareable=shareable)
if cls is None:
return wrapper
return wrapper(cls)
__all__ = ["Event", "event", "EventHandlingType", "shareable_events", "SwitchInfo"]