"""Protocols for EventChannel handlers and backends."""
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, runtime_checkable
if TYPE_CHECKING:
from collections.abc import Sequence
from sqlspec.extensions.events._models import EventMessage
__all__ = ("AsyncEventBackendProtocol", "AsyncEventHandler", "SyncEventBackendProtocol", "SyncEventHandler")
[docs]
class AsyncEventHandler(Protocol):
"""Protocol describing async event handler callables."""
[docs]
async def __call__(self, message: "EventMessage") -> Any: # pragma: no cover
"""Process a queued event message asynchronously."""
[docs]
class SyncEventHandler(Protocol):
"""Protocol describing sync event handler callables."""
[docs]
def __call__(self, message: "EventMessage") -> Any: # pragma: no cover
"""Process a queued event message synchronously."""
[docs]
@runtime_checkable
class AsyncEventBackendProtocol(Protocol):
"""Protocol for async event backends.
All async event backends (native or queue-based) must implement these methods.
"""
supports_async: ClassVar[bool]
backend_name: ClassVar[str]
[docs]
async def publish(self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None) -> str:
"""Publish an event to a channel.
Args:
channel: Target channel name.
payload: Event payload (must be JSON-serializable).
metadata: Optional metadata dict.
Returns:
The event ID.
"""
...
[docs]
async def publish_many(self, events: "Sequence[tuple[str, dict[str, Any], dict[str, Any] | None]]") -> list[str]:
"""Publish independent events as one grouped backend operation.
Implementations with native batching must preserve input order in the
returned event IDs. Backends without native batching are invoked through
the event channel's single-event fallback.
"""
...
[docs]
async def dequeue(self, channel: str, poll_interval: float) -> "EventMessage | None":
"""Dequeue an event from the channel.
Args:
channel: Channel name to listen on.
poll_interval: Timeout in seconds to wait for a notification.
Returns:
EventMessage if a notification was received, None otherwise.
"""
...
[docs]
async def ack(self, event_id: str) -> None:
"""Acknowledge an event.
Args:
event_id: ID of the event to acknowledge.
"""
...
[docs]
async def nack(self, event_id: str) -> None:
"""Return an event to the queue for redelivery.
Args:
event_id: ID of the event to return.
"""
...
[docs]
async def shutdown(self) -> None:
"""Shutdown the backend and release resources."""
...
[docs]
@runtime_checkable
class SyncEventBackendProtocol(Protocol):
"""Protocol for sync event backends.
All sync event backends (native or queue-based) must implement these methods.
"""
supports_sync: ClassVar[bool]
backend_name: ClassVar[str]
[docs]
def publish(self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None) -> str:
"""Publish an event to a channel.
Args:
channel: Target channel name.
payload: Event payload (must be JSON-serializable).
metadata: Optional metadata dict.
Returns:
The event ID.
"""
...
[docs]
def publish_many(self, events: "Sequence[tuple[str, dict[str, Any], dict[str, Any] | None]]") -> list[str]:
"""Publish independent events as one grouped backend operation.
Implementations with native batching must preserve input order in the
returned event IDs. Backends without native batching are invoked through
the event channel's single-event fallback.
"""
...
[docs]
def dequeue(self, channel: str, poll_interval: float) -> "EventMessage | None":
"""Dequeue an event from the channel.
Args:
channel: Channel name to listen on.
poll_interval: Timeout in seconds to wait for a notification.
Returns:
EventMessage if a notification was received, None otherwise.
"""
...
[docs]
def ack(self, event_id: str) -> None:
"""Acknowledge an event.
Args:
event_id: ID of the event to acknowledge.
"""
...
[docs]
def nack(self, event_id: str) -> None:
"""Return an event to the queue for redelivery.
Args:
event_id: ID of the event to return.
"""
...
[docs]
def shutdown(self) -> None:
"""Shutdown the backend and release resources."""
...
