prefect_redis.messaging

Functions

ephemeral_subscription ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L600" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

ephemeral_subscription(topic: str, source: Optional[str] = None, group: Optional[str] = None) -> AsyncGenerator[dict[str, Any], None]

break_topic ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L617" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

break_topic()

Classes

RedisMessagingPublisherSettings ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L63" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Settings for the Redis messaging publisher.

No settings are required to be set by the user but any of the settings can be overridden by the user using environment variables.

RedisMessagingConsumerSettings ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L89" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Settings for the Redis messaging consumer.

No settings are required to be set by the user but any of the settings can be overridden by the user using environment variables.

Cache ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L125" 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/integrations/prefect-redis/prefect_redis/messaging.py#L130" 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/integrations/prefect-redis/prefect_redis/messaging.py#L159" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

forget_duplicates(self, attribute: str, messages: list[M]) -> None

without_duplicates ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L133" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

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

RedisStreamsMessage ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L173" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

A message sent to a Redis stream.

Methods:

acknowledge ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L191" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

acknowledge(self) -> None

Subscription ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L196" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

A subscription-like object for Redis. We mimic the memory subscription interface so that we can set max_retries and handle dead letter queue storage in Redis.

Publisher ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L207" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Methods:

publish_data ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L263" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

publish_data(self, data: bytes, attributes: dict[str, Any])

Consumer ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L299" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Consumer implementation for Redis Streams with DLQ support.

Methods:

process_pending_messages ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L364" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

process_pending_messages(self, handler: MessageHandler, redis_client: Redis, message_batch_size: int, start_id: str = '0-0')

run ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L421" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

run(self, handler: MessageHandler) -> None