API Reference

This page is the API reference for kubemq-faststream: every exported symbol, the messaging-pattern and acknowledgement enums, the Events Store start positions, the subscriber configuration fields, and the broker.request() request-reply signature. For connection options, URL formats, and environment variables, see the configuration reference.

Public API Surface

Everything in the package's __all__ is importable directly from kubemq_faststream. The classes are loaded lazily — the package only checks that the kubemq SDK is installed when one of these symbols is first accessed.

from kubemq_faststream import (
    KubeMQBroker,
    KubeMQRouter,
    KubeMQPublisher,
    KubeMQMessage,
    KubeMQPublishCommand,
    TestKubeMQBroker,
    AckPolicy,
    StartPosition,
    KubeMQPattern,
    FeatureNotSupportedException,
    __version__,
)

KubeMQPattern

The KubeMQPattern enum names the five KubeMQ messaging patterns. You rarely reference it directly — the @broker.subscriber(...) keyword and the broker.publish(...)/broker.request(...) keyword select the pattern for you — but it appears in subscriber configuration and KubeMQPublishCommand.

KubeMQPattern is a StrEnum, so each member compares equal to its string value.

AckPolicy

AckPolicy is re-exported from faststream.middlewares and controls how queue messages are settled. It applies to the Queues pattern; for Events and Events Store (fire-and-forget) the ack/nack operations are no-ops.

from kubemq_faststream import KubeMQBroker, AckPolicy

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(queues="tasks", ack_policy=AckPolicy.ACK)
async def process_task(task: dict) -> None:
    ...  # auto-acked on success, nacked on exception

See Queues for the full settlement walkthrough.

StartPosition

StartPosition selects where an Events Store subscriber begins reading. Three of the six positions require a companion start_value (passed to @broker.subscriber(events_store=..., start_position=..., start_value=...)). start_value is typed int | float | None and must be a positive number for the positions that require it.

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(
    events_store="audit-log",
    start_position=StartPosition.START_FROM_FIRST,
)
async def on_audit(msg: dict) -> None:
    print(f"Audit event: {msg}")

See Events Store for replay walkthroughs.

Subscriber Configuration

The @broker.subscriber(...) decorator accepts a pattern keyword (events=, events_store=, queues=, commands=, or queries=) plus the fields below. The pattern keyword's value is the channel name. The configuration is backed by KubeMQSubscriberConfig.

For Events Store subscribers, also pass start_position and (where required) start_value as described in StartPosition.

broker.request() — RPC Reference

broker.request(...) performs request-reply RPC. It is valid only for the Commands and Queries patterns — calling it without commands= or queries= raises FeatureNotSupportedException. The call blocks until the handler responds or the timeout elapses, and returns the handler's response.

async def request(
    message,
    /,
    *,
    commands: str | None = None,
    queries: str | None = None,
    timeout: int | None = None,
    cache_key: str | None = None,
    cache_ttl: int | None = None,
    headers: dict[str, str] | None = None,
    metadata: str = "",
    correlation_id: str | None = None,
) -> Any: ...

@broker.subscriber(commands="example.commands.basic")
async def handle_command(msg: dict) -> dict:
    return {"status": "executed", "action": msg["action"], "success": True}

# caller
response = await broker.request(
    {"action": "create_user", "name": "Alice"},
    commands="example.commands.basic",
)

See Commands & Queries for the full walkthrough, including caching and timeout handling.

Internal Architecture

On connect, KubeMQBroker creates three KubeMQ SDK async clients, each on its own gRPC channel, sharing one set of connection settings (URL, client ID, auth token, TLS, size limits, keepalive):

Each publish, request, and @broker.subscriber call is routed to the matching client by its pattern keyword.

The clients are created when the broker connects (via start() or async with broker:) and closed on stop(). See Concepts for the lifecycle and the message pipeline.

Project Links