prefect.server.utilities.messaging

Functions

create_cache ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L169" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

create_cache() -> Cache

Creates a new cache with the applications default settings.

Returns:

• a new Cache instance

create_publisher ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L195" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

create_publisher(topic: str, cache: Optional[Cache] = None, deduplicate_by: Optional[str] = None) -> Publisher

Creates a new publisher with the applications default settings. Args: topic: the topic to publish to Returns: a new Consumer instance

ephemeral_subscription ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L213" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

ephemeral_subscription(topic: str) -> AsyncGenerator[Mapping[str, Any], Any]

Creates an ephemeral subscription to the given source, removing it when the context exits.

create_consumer ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L224" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

create_consumer(topic: str, **kwargs: Any) -> Consumer

Creates a new consumer with the applications default settings. Args: topic: the topic to consume from Returns: a new Consumer instance

Classes

Message ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L32" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

A protocol representing a message sent to a message broker.

Methods:

attributes ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L41" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

attributes(self) -> Mapping[str, Any]

data ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L38" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

data(self) -> Union[str, bytes]

Cache ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L44" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Methods:

clear_recently_seen_messages ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L46" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

clear_recently_seen_messages(self) -> None

forget_duplicates ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L54" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

forget_duplicates(self, attribute: str, messages: Iterable[Message]) -> None

without_duplicates ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L49" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

without_duplicates(self, attribute: str, messages: Iterable[M]) -> list[M]

Publisher ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L59" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Methods:

publish_data ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L68" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

publish_data(self, data: bytes, attributes: Mapping[str, str]) -> None

CapturedMessage ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L85" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

CapturingPublisher ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L90" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Methods:

publish_data ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L115" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

publish_data(self, data: bytes, attributes: Mapping[str, str]) -> None

publish_data ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L68" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

publish_data(self, data: bytes, attributes: Mapping[str, str]) -> None

StopConsumer ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L129" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Exception to raise to stop a consumer.

Consumer ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L138" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Abstract base class for consumers that receive messages from a message broker and call a handler function for each message received.

Methods:

cleanup ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L152" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

cleanup(self) -> None

Cleanup resources when the consumer is stopped.

Override this method in subclasses that need to perform cleanup, such as unsubscribing from topics or closing connections.

The default implementation is a no-op, which is appropriate for consumers that don't need explicit cleanup.

run ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L148" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

run(self, handler: MessageHandler) -> None

Runs the consumer (indefinitely)

CacheModule ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L165" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

BrokerModule ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/utilities/messaging/__init__.py#L183" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}