Shared Types

abe/types.py centralises all type aliases and protocols that define the Abstract Backend surface. Installing the package from PyPI automatically exposes these types to downstream projects thanks to the bundled py.typed marker.

JSON aliases

Alias	Definition	Notes
JSONPrimitive	`str \	int \
JSONValue	`JSONPrimitive \	JSONDict \
JSONDict	dict[str, JSONValue]	Canonical representation of JSON objects
JSONList	list[JSONValue]	Canonical representation of JSON arrays

Use these when authoring payload structures so static analysers understand the nested JSON semantics.

Event payloads and handlers

• WebhookEventPayload = dict[str, Any]
• SyncEventHandlerFunc = Callable[[WebhookEventPayload], None]
• AsyncEventHandlerFunc = Callable[[WebhookEventPayload], Awaitable[None]]
• EventHandlerFunc = SyncEventHandlerFunc | AsyncEventHandlerFunc

These aliases demonstrate how to declare handlers that satisfy EventHandlerProtocol.

Message-queue aliases

Alias	Description
MessageQueueKey	Routing key/topic identifier (string).
MessageQueuePayload	JSON-compatible dictionary representing the message.
MessageQueueMessage	Provider-defined structure for consumed messages (payload plus metadata).
MessageQueueBackendConfig	dict[str, str | int | bool] used during initialisation.
ConsumerGroup	str | None naming group-based consumers.

Protocols

EventHandlerProtocol

@runtime_checkable
class EventHandlerProtocol(Protocol):
    async def handle_event(self, event: WebhookEventPayload) -> None: ...

Any class with an async handle_event method can be treated as an event handler. This is useful when you want to inject class-based processors into higher-level orchestration code.

MessageQueueBackendProtocol

The structural protocol re-exported as MessageQueueBackend. Providers must implement:

• async publish(key: MessageQueueKey, payload: MessageQueuePayload) -> None
• async consume(*, group: ConsumerGroup = None) -> AsyncIterator[MessageQueueMessage]
• @classmethod from_env(cls) -> MessageQueueBackendProtocol

Best practices

• When creating new modules, import aliases from abe.types instead of re-declaring them.
• Export any additional public types from your provider package’s __all__ to keep MyPy happy.
• The verification script scripts/ci/verify_type_checking.sh ensures newly added types remain part of the public surface—update it if you add more names.