> ## Documentation Index
> Fetch the complete documentation index at: /docs/llms.txt
> Use this file to discover all available pages before exploring further.

# messaging

# `prefect_redis.messaging`

## Functions

### `ephemeral_subscription` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L699](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L699)</sup>

```python theme={null}
ephemeral_subscription(topic: str, source: Optional[str] = None, group: Optional[str] = None) -> AsyncGenerator[dict[str, Any], None]
```

### `break_topic` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L731](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L731)</sup>

```python theme={null}
break_topic()
```

## Classes

### `RedisMessagingPublisherSettings` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L72](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L72)</sup>

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` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L98](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L98)</sup>

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` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L166](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L166)</sup>

**Methods:**

#### `clear_recently_seen_messages` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L171](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L171)</sup>

```python theme={null}
clear_recently_seen_messages(self) -> None
```

#### `forget_duplicates` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L200](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L200)</sup>

```python theme={null}
forget_duplicates(self, attribute: str, messages: list[M]) -> None
```

#### `without_duplicates` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L174](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L174)</sup>

```python theme={null}
without_duplicates(self, attribute: str, messages: list[M]) -> list[M]
```

### `RedisStreamsMessage` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L214](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L214)</sup>

A message sent to a Redis stream.

**Methods:**

#### `acknowledge` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L232](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L232)</sup>

```python theme={null}
acknowledge(self) -> None
```

### `Subscription` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L237](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L237)</sup>

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` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L248](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L248)</sup>

**Methods:**

#### `publish_data` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L305](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L305)</sup>

```python theme={null}
publish_data(self, data: bytes, attributes: dict[str, Any])
```

### `Consumer` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L341](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L341)</sup>

Consumer implementation for Redis Streams with DLQ support.

**Methods:**

#### `process_pending_messages` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L455](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L455)</sup>

```python theme={null}
process_pending_messages(self, handler: MessageHandler, redis_client: Redis, message_batch_size: int, start_id: str = '0-0')
```

#### `run` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect\_redis/messaging.py#L512](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-redis/prefect_redis/messaging.py#L512)</sup>

```python theme={null}
run(self, handler: MessageHandler) -> None
```