`prefect.server.utilities.messaging`

Functions

`create_cache`

create_cache() -> Cache

Creates a new cache with the applications default settings. Returns:

a new Cache instance

`create_publisher`

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`

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`

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 protocol representing a message sent to a message broker. Methods:

`attributes`

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

`data`

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

`Cache`

Methods:

`clear_recently_seen_messages`

clear_recently_seen_messages(self) -> None

`forget_duplicates`

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

`without_duplicates`

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

`Publisher`

Methods:

`publish_data`

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

`CapturedMessage`

`CapturingPublisher`

Methods:

`publish_data`

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

`publish_data`

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

`StopConsumer`

Exception to raise to stop a consumer.

`Consumer`

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

`cleanup`

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`

run(self, handler: MessageHandler) -> None

Runs the consumer (indefinitely)

API Reference

init

`prefect.server.utilities.messaging`

Functions

`create_cache`

`create_publisher`

`ephemeral_subscription`

`create_consumer`

Classes

`Message`

`attributes`

`data`

`Cache`

`clear_recently_seen_messages`

`forget_duplicates`

`without_duplicates`

`Publisher`

`publish_data`

`CapturedMessage`

`CapturingPublisher`

`publish_data`

`publish_data`

`StopConsumer`

`Consumer`

`cleanup`

`run`

`CacheModule`

`BrokerModule`

​prefect.server.utilities.messaging

​Functions

​create_cache

​create_publisher

​ephemeral_subscription

​create_consumer

​Classes

​Message

​attributes

​data

​Cache

​clear_recently_seen_messages

​forget_duplicates

​without_duplicates

​Publisher

​publish_data

​CapturedMessage

​CapturingPublisher

​publish_data

​publish_data

​StopConsumer

​Consumer

​cleanup

​run

​CacheModule

​BrokerModule

`prefect.server.utilities.messaging`

Functions

`create_cache`

`create_publisher`

`ephemeral_subscription`

`create_consumer`

Classes

`Message`

`attributes`

`data`

`Cache`

`clear_recently_seen_messages`

`forget_duplicates`

`without_duplicates`

`Publisher`

`publish_data`

`CapturedMessage`

`CapturingPublisher`

`publish_data`

`publish_data`

`StopConsumer`

`Consumer`

`cleanup`

`run`

`CacheModule`

`BrokerModule`