### Install esdb-py library

Source: https://github.com/andriykohut/esdb-py/blob/main/README.md

Commands to install the esdb-py package using standard package managers.

```shell
pip install esdb
```

```shell
poetry add esdb
```

--------------------------------

### Configure ESClient connection

Source: https://github.com/andriykohut/esdb-py/blob/main/README.md

Examples of initializing the ESClient using different connection strings for discovery and security configurations.

```python
from esdb import ESClient

# Discovery with node preference
client = ESClient("esdb+discover://admin:changeit@localhost:2111?nodePreference=follower")

# Connect without TLS
client_insecure = ESClient("esdb://localhost:2111?tls=false")

# Secure connection with basic auth and keepalive
client_secure = ESClient("esdb://admin:changeit@localhost:2111?tlsCafile=certs/ca/ca.crt&keepAliveInterval=5&keepAliveTimeout=5")
```

--------------------------------

### Create Stream Subscription - esdb-py

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Creates a persistent subscription to a specific stream with configurable settings. It can be configured to start from the beginning or from the end of the stream.

```python
import asyncio
from esdb import ESClient
from esdb.subscriptions import SubscriptionSettings

client = ESClient("esdb://localhost:2111?tls=false")

async def create_subscription():
    async with client.connect() as conn:
        # Create a persistent subscription to a stream
        await conn.subscriptions.create_stream_subscription(
            stream="orders",
            group_name="order-processors",
            settings=SubscriptionSettings(
                live_buffer_size=100,
                read_batch_size=20,
                history_buffer_size=100,
                checkpoint_ms=10000,  # Checkpoint every 10 seconds
                max_retry_count=5,
                max_subscriber_count=10,
                message_timeout_ms=30000,  # 30 second message timeout
                consumer_strategy=SubscriptionSettings.ConsumerStrategy.ROUND_ROBIN,
            ),
        )
        print("Subscription created successfully")

        # Create subscription starting from the end (new events only)
        await conn.subscriptions.create_stream_subscription(
            stream="notifications",
            group_name="notification-handlers",
            settings=SubscriptionSettings(
                live_buffer_size=50,
                read_batch_size=10,
                history_buffer_size=50,
                checkpoint_ms=5000,
            ),
            backwards=True,  # Start from end of stream
        )

asyncio.run(create_subscription())
```

--------------------------------

### Handle EventStoreDB Exceptions

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Provides examples of catching specific library exceptions such as StreamNotFound and WrongExpectedVersion to handle stream state conflicts and missing resources.

```python
import asyncio
from esdb import ESClient
from esdb.exceptions import ClientException, WrongExpectedVersion, StreamNotFound
from esdb.streams import StreamState

client = ESClient("esdb://localhost:2111?tls=false")

async def handle_errors():
    async with client.connect() as conn:
        try:
            async for event in conn.streams.read(stream="nonexistent", count=10):
                print(event)
        except StreamNotFound as err:
            print(f"Stream error: {err}")
        try:
            await conn.streams.append(stream="user-123", event_type="UserUpdated", data={"name": "Jane"}, stream_state=StreamState.NO_STREAM)
        except WrongExpectedVersion as err:
            print(f"Concurrency error: {err}")
        try:
            await conn.streams.delete(stream="protected-stream")
        except ClientException as err:
            print(f"Client error: {err}")
```

--------------------------------

### Create Stream Subscription

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Creates a persistent subscription to a specific stream with configurable settings for consumer groups. Supports starting from the end of the stream.

```APIDOC
## POST /streams/{streamName}/subscriptions/{groupName}

### Description
Creates a persistent subscription to a specific stream for a given consumer group. This allows for reliable, at-least-once processing of events from the stream.

### Method
POST

### Endpoint
/streams/{streamName}/subscriptions/{groupName}

### Parameters
#### Path Parameters
- **streamName** (string) - Required - The name of the stream to subscribe to.
- **groupName** (string) - Required - The name of the consumer group.

#### Query Parameters
- **backwards** (boolean) - Optional - If true, the subscription starts from the end of the stream, processing only new events.

#### Request Body
- **settings** (SubscriptionSettings) - Required - Configuration settings for the subscription.
  - **live_buffer_size** (integer) - Optional - The size of the buffer for live messages.
  - **read_batch_size** (integer) - Optional - The number of messages to read in each batch.
  - **history_buffer_size** (integer) - Optional - The size of the buffer for historical messages.
  - **checkpoint_ms** (integer) - Optional - The interval in milliseconds for checkpointing.
  - **max_retry_count** (integer) - Optional - The maximum number of retries for failed messages.
  - **max_subscriber_count** (integer) - Optional - The maximum number of concurrent subscribers.
  - **message_timeout_ms** (integer) - Optional - The timeout in milliseconds for processing a message.
  - **consumer_strategy** (string) - Optional - The strategy for distributing messages among subscribers (e.g., ROUND_ROBIN).

### Request Example
```json
{
  "settings": {
    "live_buffer_size": 100,
    "read_batch_size": 20,
    "history_buffer_size": 100,
    "checkpoint_ms": 10000,
    "max_retry_count": 5,
    "max_subscriber_count": 10,
    "message_timeout_ms": 30000,
    "consumer_strategy": "ROUND_ROBIN"
  }
}
```

### Response
#### Success Response (200)
- **Subscription created successfully** (string) - Confirmation message.

#### Response Example
```json
"Subscription created successfully"
```
```

--------------------------------

### Get Persistent Subscription Info - Python

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Retrieves detailed information about a persistent subscription, including connection details, statistics like total items processed, average per second, buffer counts, and parked messages. It supports fetching info for both stream-specific and all-stream subscriptions.

```python
import asyncio
from esdb import ESClient

client = ESClient("esdb://localhost:2111?tls=false")

async def get_subscription_info():
    async with client.connect() as conn:
        # Get info for a stream subscription
        info = await conn.subscriptions.get_info(
            group_name="order-processors",
            stream="orders",
        )
        print(f"Subscription: {info.group_name}")
        print(f"Status: {info.status}")
        print(f"Total items processed: {info.total_items}")
        print(f"Average per second: {info.average_per_second}")
        print(f"Live buffer count: {info.live_buffer_count}")
        print(f"Parked messages: {info.parked_message_count}")
        print(f"Consumer strategy: {info.consumer_strategy}")
        print(f"Max subscribers: {info.max_subscriber_count}")
        print(f"Connections: {len(info.connections)}")
        for conn_info in info.connections:
            print(f"  - {conn_info.connection_name}: {conn_info.total_items} items")

        # Get info for an all-stream subscription
        all_info = await conn.subscriptions.get_info(
            group_name="analytics-processors",
            stream=None,  # $all subscription
        )
        print(f"All-stream subscription total: {all_info.total_items}")

asyncio.run(get_subscription_info())
```

--------------------------------

### GET /streams/read

Source: https://github.com/andriykohut/esdb-py/blob/main/README.md

Reads events from a stream, supporting forward/backward reading and subscriptions.

```APIDOC
## GET /streams/read

### Description
Reads events from a stream. Supports reading backwards, starting from a specific revision, and establishing a catch-up subscription.

### Method
GET

### Parameters
#### Query Parameters
- **stream** (string) - Required - The name of the stream to read.
- **count** (integer) - Optional - Number of events to read.
- **backwards** (boolean) - Optional - If true, reads events in reverse order.
- **revision** (integer) - Optional - The starting revision for the read.
- **subscribe** (boolean) - Optional - If true, initiates a catch-up subscription.

### Response
#### Success Response (200)
- **result** (stream) - An asynchronous stream of event objects containing the event data.
```

--------------------------------

### Read Events from Stream

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Reads events from a specified stream. Supports forward or backward reading, starting from specific revisions, and limiting the result count.

```python
import asyncio
from esdb import ESClient

client = ESClient("esdb://localhost:2111?tls=false")

async def read_stream():
    async with client.connect() as conn:
        async for event in conn.streams.read(stream="user-123", count=10):
            print(f"Event: {event.event_type}, Data: {event.data}")

        async for event in conn.streams.read(stream="user-123", count=10, backwards=True):
            print(f"Recent: {event.event_type}")

        async for event in conn.streams.read(stream="user-123", count=5, revision=3):
            print(f"From rev 3: {event.event_type}")

        async for event in conn.streams.read(stream="user-123", count=5, backwards=True, revision=10):
            print(f"Before rev 10: {event.event_type}")

        events = [e async for e in conn.streams.read(stream="user-123", count=100)]
        print(f"Total events: {len(events)}")

asyncio.run(read_stream())
```

--------------------------------

### Connect to EventStoreDB using ESClient

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Demonstrates various ways to establish a connection to EventStoreDB using the ESClient class. It covers insecure, secure, and DNS-discovered connections, along with using the async context manager.

```python
from esdb import ESClient

# Single-node insecure connection
client = ESClient("esdb://localhost:2111?tls=false")

# Secure connection with basic auth and CA certificate
client = ESClient(
    "esdb://admin:changeit@localhost:2111?tlsCafile=certs/ca/ca.crt&keepAliveInterval=5&keepAliveTimeout=5"
)

# DNS discovery with node preference (leader, follower, or readonlyreplica)
client = ESClient(
    "esdb+discover://admin:changeit@localhost:2111?nodePreference=follower&maxDiscoverAttempts=3"
)

# Using the async context manager to connect
async def main():
    async with client.connect() as conn:
        # conn.streams - for stream operations
        # conn.subscriptions - for persistent subscriptions
        # conn.gossip - for cluster information
        pass
```

--------------------------------

### Perform stream operations and subscriptions

Source: https://github.com/andriykohut/esdb-py/blob/main/README.md

Demonstrates how to append events to a stream, read events with various filters, and initiate a catch-up subscription using the ESClient.

```python
import asyncio
import datetime
import uuid
from esdb import ESClient

client = ESClient("esdb+discover://admin:changeit@localhost:2111")
stream = f"test-{str(uuid.uuid4())}"

async def streams():
    async with client.connect() as conn:
        # Appending to stream
        for i in range(10):
            await conn.streams.append(
                stream=stream,
                event_type="test_event",
                data={"i": i, "ts": datetime.datetime.utcnow().isoformat()},
            )

        # Read events
        async for result in conn.streams.read(stream=stream, count=10):
            print(result.data)

        # Catch-up subscription
        async for result in conn.streams.read(stream=stream, subscribe=True):
            print(result.data)

asyncio.run(streams())
```

--------------------------------

### Read All Events with Filtering in Python

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Demonstrates reading events from all streams using the ESClient. It shows how to apply regex filters for event types or stream names and how to initiate real-time subscriptions with checkpointing.

```python
import asyncio
from esdb import ESClient
from esdb.shared import Filter

client = ESClient("esdb://localhost:2111?tls=false")

async def read_all_with_filters():
    async with client.connect() as conn:
        # Read all events (first 100)
        async for event in conn.streams.read_all(count=100):
            print(f"[{event.stream_name}] {event.event_type}: {event.data}")

        # Filter by event type using regex
        async for event in conn.streams.read_all(
            count=50,
            filter_by=Filter(
                kind=Filter.Kind.EVENT_TYPE,
                regex="^User",  # Events starting with "User"
            ),
        ):
            print(f"User event: {event.event_type}")

        # Filter by stream name using regex
        async for event in conn.streams.read_all(
            count=50,
            filter_by=Filter(
                kind=Filter.Kind.STREAM,
                regex="^order-",  # Streams starting with "order-"
            ),
        ):
            print(f"Order stream: {event.stream_name} - {event.event_type}")

        # Subscribe to all events with filtering (real-time)
        async for event in conn.streams.read_all(
            subscribe=True,
            filter_by=Filter(
                kind=Filter.Kind.EVENT_TYPE,
                regex="^Payment",
                checkpoint_interval_multiplier=1000,  # Required for subscriptions
            ),
        ):
            print(f"Payment event received: {event.event_type}")

asyncio.run(read_all_with_filters())
```

--------------------------------

### Create Catch-up Subscription

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Establishes a real-time subscription to a stream. It processes all existing events first and then continues to receive new events as they are appended.

```python
import asyncio
from esdb import ESClient

client = ESClient("esdb://localhost:2111?tls=false")

async def catchup_subscription():
    async with client.connect() as conn:
        async for event in conn.streams.read(stream="user-123", subscribe=True):
            print(f"Received: {event.event_type} - {event.data}")

            if event.event_type == "UserCreated":
                print(f"New user: {event.data.get('name')}")
            elif event.event_type == "UserDeleted":
                print("User was deleted, stopping subscription")
                break

asyncio.run(catchup_subscription())
```

--------------------------------

### Update Persistent Subscription Settings

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Demonstrates how to update configuration settings for both stream-specific and all-stream persistent subscriptions using SubscriptionSettings.

```python
import asyncio
from esdb import ESClient
from esdb.subscriptions import SubscriptionSettings

client = ESClient("esdb://localhost:2111?tls=false")

async def update_subscription():
    async with client.connect() as conn:
        await conn.subscriptions.update_stream_subscription(
            stream="orders",
            group_name="order-processors",
            settings=SubscriptionSettings(
                live_buffer_size=200,
                read_batch_size=40,
                history_buffer_size=200,
                checkpoint_ms=5000,
                max_retry_count=10,
                max_subscriber_count=20,
            ),
        )
        await conn.subscriptions.update_all_subscription(
            group_name="analytics-processors",
            settings=SubscriptionSettings(
                live_buffer_size=500,
                read_batch_size=100,
                history_buffer_size=500,
                checkpoint_ms=30000,
            ),
        )

asyncio.run(update_subscription())
```

--------------------------------

### Catch-up subscription with event filtering

Source: https://github.com/andriykohut/esdb-py/blob/main/README.md

Shows how to read events from all streams using a catch-up subscription. It utilizes a filter to only process events matching a specific regex pattern.

```python
import uuid
import asyncio

from esdb import ESClient
from esdb.shared import Filter


async def filters():
    async with ESClient("esdb+discover://admin:changeit@localhost:2111").connect() as conn:
        for i in range(10):
            await conn.streams.append(stream=str(uuid.uuid4()), event_type=f"prefix-{i}", data=b"")
        async for event in conn.streams.read_all(
                subscribe=True,
                filter_by=Filter(
                    kind=Filter.Kind.EVENT_TYPE,
                    regex="^prefix-",
                    checkpoint_interval_multiplier=1000,
                ),
        ):
            print(event)


asyncio.run(filters())
```

--------------------------------

### Handle Discovery Errors in Python

Source: https://context7.com/andriykohut/esdb-py/llms.txt

This snippet demonstrates how to wrap an EventStoreDB client connection attempt in a try-except block to catch DiscoveryError. It is useful for gracefully handling scenarios where the client cannot resolve the cluster nodes after a specified number of attempts.

```python
async def handle_discovery():
    try:
        bad_client = ESClient("esdb+discover://admin:wrong@invalid-host:2111?maxDiscoverAttempts=2")
        async with bad_client.connect() as conn:
            pass
    except DiscoveryError as err:
        print(f"Discovery failed: {err}")

asyncio.run(handle_discovery())
```

--------------------------------

### Batch append events to a stream

Source: https://github.com/andriykohut/esdb-py/blob/main/README.md

Demonstrates how to append multiple events to a single stream in one batch operation. Note that this feature requires EventStoreDB v21.10 or later.

```python
import asyncio
import uuid

from esdb import ESClient
from esdb.streams import Message


async def batch_append():
    stream = str(uuid.uuid4())
    messages: list[Message] = [
        Message(event_type="one", data={"item": 1}),
        Message(event_type="one", data={"item": 2}),
        Message(event_type="one", data={"item": 3}),
        Message(event_type="two", data={"item": 1}),
        Message(event_type="two", data={"item": 2}),
        Message(event_type="two", data={"item": 3}),
    ]
    async with ESClient("esdb+discover://admin:changeit@localhost:2111").connect() as conn:
        response = await conn.streams.batch_append(stream=stream, messages=messages)
        assert response.current_revision == 5
        events = [e async for e in conn.streams.read(stream=stream, count=50)]
        assert len(events) == 6


asyncio.run(batch_append())
```

--------------------------------

### Manage persistent subscriptions

Source: https://github.com/andriykohut/esdb-py/blob/main/README.md

Covers the lifecycle of persistent subscriptions, including creation with custom settings, filtering, event acknowledgment (ACK), negative acknowledgment (NACK), and deletion.

```python
import asyncio
from esdb import ESClient
from esdb.shared import Filter
from esdb.subscriptions import SubscriptionSettings, NackAction

client = ESClient("esdb+discover://admin:changeit@localhost:2111")

stream = "stream-foo"
group = "group-bar"


async def persistent():
    async with client.connect() as conn:
        for i in range(50):
            await conn.streams.append(stream, "foobar", {"i": i})

        await conn.subscriptions.create_stream_subscription(
            stream=stream,
            group_name=group,
            settings=SubscriptionSettings(
                max_subscriber_count=50,
                read_batch_size=5,
                live_buffer_size=10,
                history_buffer_size=10,
                consumer_strategy=SubscriptionSettings.ConsumerStrategy.ROUND_ROBIN,
                checkpoint_ms=10000,
            ),
        )

        await conn.subscriptions.create_all_subscription(
            group_name="subscription_group",
            filter_by=Filter(kind=Filter.Kind.EVENT_TYPE, regex="^some_type$", checkpoint_interval_multiplier=200),
            settings=SubscriptionSettings(
                read_batch_size=50,
                live_buffer_size=100,
                history_buffer_size=100,
                max_retry_count=2,
                checkpoint_ms=20000,
            ),
        )

    async with client.connect() as conn:
        sub = conn.subscriptions.subscribe(stream=stream, group_name=group, buffer_size=5)
        async for event in sub:
            try:
                print(event)
                await sub.ack([event])
            except Exception as err:
                await sub.nack([event], NackAction.RETRY, reason=str(err))

        info = await conn.subscriptions.get_info(group, stream)
        assert info.group_name == group

        await conn.subscriptions.delete(group, stream)
        
        subs = await conn.subscriptions.list()
        for sub in subs:
            print(sub.total_items)


asyncio.run(persistent())
```

--------------------------------

### Subscribe to Persistent Subscription - Python

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Reads events from a persistent subscription, supporting acknowledgment and negative acknowledgment. It processes events, acknowledges successful ones, and handles retries or parking for failed events. It also shows how to subscribe to the all-stream.

```python
import asyncio
from esdb import ESClient
from esdb.subscriptions import NackAction

client = ESClient("esdb://localhost:2111?tls=false")

async def process_subscription():
    async with client.connect() as conn:
        # Subscribe to a persistent subscription
        sub = conn.subscriptions.subscribe(
            stream="orders",
            group_name="order-processors",
            buffer_size=10,  # Prefetch buffer
        )

        async for event in sub:
            print(f"Processing: {event.type} (retry: {event.retry_count})")
            print(f"  Stream: {event.stream}, ID: {event.id}")
            print(f"  Data: {event.data}")

            try:
                # Process the event
                if event.type == "OrderCreated":
                    order_data = event.data
                    # ... process order ...
                    print(f"Processed order: {order_data}")

                # Acknowledge successful processing
                await sub.ack([event])
                print("Event acknowledged")

            except ValueError as err:
                # Retry the event (will be redelivered)
                await sub.nack([event], NackAction.RETRY, reason=str(err))
                print(f"Event will be retried: {err}")

            except Exception as err:
                # Park the event (move to parked queue for manual inspection)
                await sub.nack([event], NackAction.PARK, reason=str(err))
                print(f"Event parked: {err}")

        # Subscribe to all-stream subscription (no stream parameter)
        all_sub = conn.subscriptions.subscribe(
            group_name="analytics-processors",
            buffer_size=20,
            stream=None,  # Subscribe to $all
        )
        async for event in all_sub:
            await all_sub.ack([event])

asyncio.run(process_subscription())
```

--------------------------------

### Create All-Stream Subscription with Filtering - esdb-py

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Creates a persistent subscription to all streams, optionally filtering events by type or stream name using regular expressions. Requires EventStoreDB v21.10+.

```python
import asyncio
from esdb import ESClient
from esdb.shared import Filter
from esdb.subscriptions import SubscriptionSettings

client = ESClient("esdb://localhost:2111?tls=false")

async def create_all_subscription():
    async with client.connect() as conn:
        # Create subscription to all events matching a filter
        await conn.subscriptions.create_all_subscription(
            group_name="analytics-processors",
            settings=SubscriptionSettings(
                live_buffer_size=200,
                read_batch_size=50,
                history_buffer_size=200,
                checkpoint_ms=20000,
                max_retry_count=3,
            ),
            filter_by=Filter(
                kind=Filter.Kind.EVENT_TYPE,
                regex="^Analytics",  # Only analytics events
                checkpoint_interval_multiplier=200,
            ),
        )
        print("All-stream subscription created")

        # Filter by stream name pattern
        await conn.subscriptions.create_all_subscription(
            group_name="user-event-processors",
            settings=SubscriptionSettings(
                live_buffer_size=100,
                read_batch_size=20,
                history_buffer_size=100,
                checkpoint_ms=10000,
            ),
            filter_by=Filter(
                kind=Filter.Kind.STREAM,
                regex="^user-",  # All user streams
                checkpoint_interval_multiplier=100,
            ),
        )

asyncio.run(create_all_subscription())
```

--------------------------------

### PersistentSubscriptions.get_info

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Retrieves detailed information about a persistent subscription, including connection info and statistics.

```APIDOC
## GET /subscriptions/info

### Description
Retrieves detailed information about a persistent subscription, including connection info and statistics.

### Method
GET

### Endpoint
/subscriptions/info

### Parameters
#### Query Parameters
- **stream** (string) - Required - The name of the stream the subscription is for, or None for the all-stream.
- **group_name** (string) - Required - The name of the persistent subscription group.

### Request Body
(Not applicable for this method)

### Response
#### Success Response (200)
- **group_name** (string) - The name of the subscription group.
- **status** (string) - The current status of the subscription.
- **total_items** (integer) - Total number of items processed by the subscription.
- **average_per_second** (float) - Average number of items processed per second.
- **live_buffer_count** (integer) - Number of items currently in the live buffer.
- **parked_message_count** (integer) - Number of messages parked.
- **consumer_strategy** (string) - The consumer strategy used by the subscription.
- **max_subscriber_count** (integer) - Maximum number of concurrent subscribers.
- **connections** (array) - List of connection information.
  - **connection_name** (string) - Name of the connection.
  - **total_items** (integer) - Total items processed by this connection.

#### Response Example
```json
{
  "group_name": "order-processors",
  "status": "Live",
  "total_items": 1500,
  "average_per_second": 10.5,
  "live_buffer_count": 50,
  "parked_message_count": 5,
  "consumer_strategy": "DispatchToSingle",
  "max_subscriber_count": 10,
  "connections": [
    {
      "connection_name": "conn-1",
      "total_items": 700
    },
    {
      "connection_name": "conn-2",
      "total_items": 800
    }
  ]
}
```
```

--------------------------------

### Create All-Stream Subscription

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Creates a persistent subscription to all streams with optional filtering. Requires EventStoreDB v21.10+.

```APIDOC
## POST /streams/$all/subscriptions/{groupName}

### Description
Creates a persistent subscription to all streams. This endpoint allows for subscribing to all events across the entire database, with optional filtering based on event type or stream name patterns.

### Method
POST

### Endpoint
/streams/$all/subscriptions/{groupName}

### Parameters
#### Path Parameters
- **groupName** (string) - Required - The name of the consumer group.

#### Query Parameters
- **filter_by** (Filter) - Optional - Specifies the filtering criteria for the subscription.
  - **kind** (string) - Required - The type of filter (e.g., EVENT_TYPE, STREAM).
  - **regex** (string) - Required - The regular expression to match against.
  - **checkpoint_interval_multiplier** (integer) - Optional - Multiplier for checkpoint interval.

#### Request Body
- **settings** (SubscriptionSettings) - Required - Configuration settings for the subscription.
  - **live_buffer_size** (integer) - Optional - The size of the buffer for live messages.
  - **read_batch_size** (integer) - Optional - The number of messages to read in each batch.
  - **history_buffer_size** (integer) - Optional - The size of the buffer for historical messages.
  - **checkpoint_ms** (integer) - Optional - The interval in milliseconds for checkpointing.
  - **max_retry_count** (integer) - Optional - The maximum number of retries for failed messages.

### Request Example
```json
{
  "settings": {
    "live_buffer_size": 200,
    "read_batch_size": 50,
    "history_buffer_size": 200,
    "checkpoint_ms": 20000,
    "max_retry_count": 3
  },
  "filter_by": {
    "kind": "EVENT_TYPE",
    "regex": "^Analytics",
    "checkpoint_interval_multiplier": 200
  }
}
```

### Response
#### Success Response (200)
- **All-stream subscription created** (string) - Confirmation message.

#### Response Example
```json
"All-stream subscription created"
```
```

--------------------------------

### List Persistent Subscriptions - Python

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Lists all persistent subscriptions available in the EventStoreDB, or filters them to show only subscriptions for a specific stream. The output includes the group name, event source (stream name), status, and total items processed for each subscription.

```python
import asyncio
from esdb import ESClient

client = ESClient("esdb://localhost:2111?tls=false")

async def list_subscriptions():
    async with client.connect() as conn:
        # List all subscriptions
        all_subs = await conn.subscriptions.list()
        print(f"Total subscriptions: {len(all_subs)}")
        for sub in all_subs:
            print(f"  {sub.group_name} -> {sub.event_source}")
            print(f"    Status: {sub.status}, Total: {sub.total_items}")

        # List subscriptions for a specific stream
        stream_subs = await conn.subscriptions.list(stream="orders")
        print(f"\nSubscriptions for 'orders' stream: {len(stream_subs)}")
        for sub in stream_subs:
            print(f"  Group: {sub.group_name}")

asyncio.run(list_subscriptions())
```

--------------------------------

### Configure Event and Stream Filtering

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Utilizes the Filter class to define regex-based rules for event types or stream names, including optional checkpoint intervals and prefix lists.

```python
from esdb.shared import Filter

event_filter = Filter(kind=Filter.Kind.EVENT_TYPE, regex="^Order")
stream_filter = Filter(kind=Filter.Kind.STREAM, regex="^user-[0-9]+$")
subscription_filter = Filter(kind=Filter.Kind.EVENT_TYPE, regex="^Payment", checkpoint_interval_multiplier=1000)
prefix_filter = Filter(kind=Filter.Kind.EVENT_TYPE, regex=".*", prefixes=["Order", "Payment", "Shipping"])
```

--------------------------------

### PersistentSubscriptions.list

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Lists all persistent subscriptions or subscriptions for a specific stream.

```APIDOC
## GET /subscriptions/list

### Description
Lists all persistent subscriptions or subscriptions for a specific stream.

### Method
GET

### Endpoint
/subscriptions/list

### Parameters
#### Query Parameters
- **stream** (string) - Optional - The name of the stream to filter subscriptions by. If omitted, lists all subscriptions.

### Request Body
(Not applicable for this method)

### Response
#### Success Response (200)
- **subscriptions** (array) - A list of subscription objects.
  - **group_name** (string) - The name of the subscription group.
  - **event_source** (string) - The stream the subscription is associated with.
  - **status** (string) - The current status of the subscription.
  - **total_items** (integer) - Total number of items processed by the subscription.

#### Response Example
```json
[
  {
    "group_name": "order-processors",
    "event_source": "orders",
    "status": "Live",
    "total_items": 1500
  },
  {
    "group_name": "analytics-processors",
    "event_source": "$all",
    "status": "Live",
    "total_items": 10000
  }
]
```
```

--------------------------------

### POST /streams/append

Source: https://github.com/andriykohut/esdb-py/blob/main/README.md

Appends events to a specific stream in EventStoreDB.

```APIDOC
## POST /streams/append

### Description
Appends one or more events to a specified stream. The stream is created if it does not exist.

### Method
POST

### Parameters
#### Request Body
- **stream** (string) - Required - The name of the stream to append to.
- **event_type** (string) - Required - The type of the event.
- **data** (object) - Required - The event payload data.

### Request Example
```python
await conn.streams.append(
    stream="test-stream",
    event_type="test_event",
    data={"i": 1, "ts": "2023-10-27T10:00:00"}
)
```

### Response
#### Success Response (200)
- **append_result** (object) - Contains the result of the append operation, including the new stream revision.
```

--------------------------------

### Streams.read_all - Read All Events with Filtering

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Reads events from all streams with optional filtering by stream name or event type using regex patterns. Supports real-time subscriptions.

```APIDOC
## Streams.read_all - Read All Events with Filtering

### Description
Reads events from all streams with optional filtering by stream name or event type using regex patterns.

### Method
GET (Implicit, via client method)

### Endpoint
Not directly applicable, as this is a client-side operation interacting with the EventStoreDB cluster.

### Parameters
#### Query Parameters
- **count** (int) - Optional - The maximum number of events to read.
- **filter_by** (Filter) - Optional - An object specifying filtering criteria (kind: EVENT_TYPE or STREAM, regex).
- **subscribe** (bool) - Optional - If true, subscribes to new events in real-time.
- **checkpoint_interval_multiplier** (int) - Optional - Required when `subscribe` is true and filtering is applied. Specifies the interval for checkpointing.

### Request Example
```python
# Example: Read first 100 events
async for event in conn.streams.read_all(count=100):
    print(f"[{{event.stream_name}}] {{event.event_type}}: {{event.data}}")

# Example: Filter by event type starting with 'User'
async for event in conn.streams.read_all(
    count=50,
    filter_by=Filter(
        kind=Filter.Kind.EVENT_TYPE,
        regex="^User",
    ),
):
    print(f"User event: {{event.event_type}}")

# Example: Filter by stream name starting with 'order-'
async for event in conn.streams.read_all(
    count=50,
    filter_by=Filter(
        kind=Filter.Kind.STREAM,
        regex="^order-",
    ),
):
    print(f"Order stream: {{event.stream_name}} - {{event.event_type}}")

# Example: Subscribe to 'Payment' events
async for event in conn.streams.read_all(
    subscribe=True,
    filter_by=Filter(
        kind=Filter.Kind.EVENT_TYPE,
        regex="^Payment",
        checkpoint_interval_multiplier=1000,
    ),
):
    print(f"Payment event received: {{event.event_type}}")
```

### Response
#### Success Response (200 OK)
- **event** (object) - An event object containing stream name, event type, data, etc.

#### Response Example
```json
{
  "id": "...",
  "stream_name": "my-stream",
  "event_type": "MyEventType",
  "data": {
    "key": "value"
  },
  "metadata": {},
  "created": "2023-10-27T10:00:00Z"
}
```
```

--------------------------------

### Append Single Event to a Stream using Streams.append

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Shows how to append a single event to an EventStoreDB stream using the `streams.append` method. It supports JSON or binary data, custom metadata, and optimistic concurrency control.

```python
import asyncio
import datetime
from esdb import ESClient
from esdb.streams import StreamState

client = ESClient("esdb://localhost:2111?tls=false")

async def append_events():
    async with client.connect() as conn:
        # Simple append with JSON data
        result = await conn.streams.append(
            stream="user-123",
            event_type="UserCreated",
            data={"name": "John Doe", "email": "john@example.com", "ts": datetime.datetime.utcnow().isoformat()},
        )
        print(f"Appended at revision: {result.current_revision}")
        # Output: Appended at revision: 0

        # Append with custom metadata
        result = await conn.streams.append(
            stream="user-123",
            event_type="UserUpdated",
            data={"email": "john.doe@example.com"},
            custom_metadata={"correlation_id": "abc-123", "caused_by": "UserCreated"},
        )

        # Append with optimistic concurrency - expect stream exists
        result = await conn.streams.append(
            stream="user-123",
            event_type="UserDeleted",
            data={"reason": "User requested deletion"},
            stream_state=StreamState.STREAM_EXISTS,
        )

        # Append at specific revision
        result = await conn.streams.append(
            stream="user-123",
            event_type="UserRestored",
            data={},
            revision=2,  # Will fail if current revision is not 2
        )

        # Append binary data
        result = await conn.streams.append(
            stream="files-456",
            event_type="FileUploaded",
            data=b"\x89PNG\r\n\x1a\n...",  # Binary content
        )

asyncio.run(append_events())
```

--------------------------------

### PersistentSubscriptions.subscribe

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Reads events from a persistent subscription with acknowledgment and negative acknowledgment support. Supports subscribing to a specific stream or the all-stream.

```APIDOC
## POST /subscriptions/subscribe

### Description
Reads events from a persistent subscription with acknowledgment and negative acknowledgment support.

### Method
POST

### Endpoint
/subscriptions/subscribe

### Parameters
#### Query Parameters
- **stream** (string) - Required - The name of the stream to subscribe to, or None for the all-stream.
- **group_name** (string) - Required - The name of the persistent subscription group.
- **buffer_size** (integer) - Optional - The prefetch buffer size for events.

### Request Body
(Not applicable for this method, parameters are passed via query)

### Response
#### Success Response (200)
- **event** (object) - An event object from the subscription.

#### Response Example
(Streaming events, no single JSON response example)

### Error Handling
- **ValueError**: Raised when an event processing fails and needs to be retried.
- **Exception**: Raised for other processing errors, potentially leading to event parking.
```

--------------------------------

### Soft Delete Stream in Python

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Performs a soft delete on a stream using the ESClient. Soft deletion allows the stream to be recreated later by appending new events, and supports optimistic concurrency via revision checks.

```python
import asyncio
from esdb import ESClient
from esdb.streams import StreamState

client = ESClient("esdb://localhost:2111?tls=false")

async def delete_stream():
    async with client.connect() as conn:
        # Simple delete
        result = await conn.streams.delete(stream="temp-stream")
        if result:
            print(f"Deleted at position: {result.commit_position}")

        # Delete with stream state expectation
        result = await conn.streams.delete(
            stream="user-old",
            stream_state=StreamState.STREAM_EXISTS,
        )

        # Delete at specific revision (optimistic concurrency)
        result = await conn.streams.delete(
            stream="user-archive",
            revision=10,  # Fails if current revision != 10
        )

asyncio.run(delete_stream())
```

--------------------------------

### Delete Persistent Subscription

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Shows how to remove persistent subscriptions from a specific stream or the $all stream by providing the group name and stream identifier.

```python
import asyncio
from esdb import ESClient

client = ESClient("esdb://localhost:2111?tls=false")

async def delete_subscription():
    async with client.connect() as conn:
        await conn.subscriptions.delete(group_name="order-processors", stream="orders")
        await conn.subscriptions.delete(group_name="analytics-processors", stream=None)

asyncio.run(delete_subscription())
```

--------------------------------

### Streams.tombstone - Permanently Delete Stream

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Permanently deletes a stream (tombstone). The stream name cannot be reused after tombstoning.

```APIDOC
## Streams.tombstone - Permanently Delete Stream

### Description
Permanently deletes a stream (tombstone). The stream name cannot be reused after tombstoning.

### Method
DELETE (with tombstone action)

### Endpoint
Not directly applicable, as this is a client-side operation interacting with the EventStoreDB cluster.

### Parameters
#### Path Parameters
- **stream** (string) - Required - The name of the stream to tombstone.

#### Query Parameters
- **stream_state** (StreamState) - Optional - Expected state of the stream (e.g., `StreamState.STREAM_EXISTS`).
- **revision** (int) - Optional - Expected revision of the stream for optimistic concurrency control.

### Request Example
```python
# Tombstone a stream permanently
result = await conn.streams.tombstone(stream="user-deleted-forever")

# Tombstone with state expectation
result = await conn.streams.tombstone(
    stream="temp-data",
    stream_state=StreamState.STREAM_EXISTS,
)

# Tombstone at specific revision
result = await conn.streams.tombstone(
    stream="old-archive",
    revision=5,
)
```

### Response
#### Success Response (200 OK)
- **commit_position** (int) - The commit position of the tombstone operation.

#### Response Example
```json
{
  "commit_position": 67890
}
```
```

--------------------------------

### Streams.delete - Soft Delete Stream

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Performs a soft delete on a stream. The stream can be recreated by appending new events.

```APIDOC
## Streams.delete - Soft Delete Stream

### Description
Performs a soft delete on a stream. The stream can be recreated by appending new events.

### Method
DELETE

### Endpoint
Not directly applicable, as this is a client-side operation interacting with the EventStoreDB cluster.

### Parameters
#### Path Parameters
- **stream** (string) - Required - The name of the stream to delete.

#### Query Parameters
- **stream_state** (StreamState) - Optional - Expected state of the stream (e.g., `StreamState.STREAM_EXISTS`).
- **revision** (int) - Optional - Expected revision of the stream for optimistic concurrency control.

### Request Example
```python
# Simple delete
result = await conn.streams.delete(stream="temp-stream")

# Delete with stream state expectation
result = await conn.streams.delete(
    stream="user-old",
    stream_state=StreamState.STREAM_EXISTS,
)

# Delete at specific revision (optimistic concurrency)
result = await conn.streams.delete(
    stream="user-archive",
    revision=10,  # Fails if current revision != 10
)
```

### Response
#### Success Response (200 OK)
- **commit_position** (int) - The commit position of the delete operation.

#### Response Example
```json
{
  "commit_position": 12345
}
```
```

--------------------------------

### Permanently Delete (Tombstone) Stream in Python

Source: https://context7.com/andriykohut/esdb-py/llms.txt

Permanently removes a stream from EventStoreDB. Once a stream is tombstoned, the stream name cannot be reused, and the operation can be guarded by stream state or revision checks.

```python
import asyncio
from esdb import ESClient
from esdb.streams import StreamState

client = ESClient("esdb://localhost:2111?tls=false")

async def tombstone_stream():
    async with client.connect() as conn:
        # Tombstone a stream permanently
        result = await conn.streams.tombstone(stream="user-deleted-forever")
        if result:
            print(f"Tombstoned at: {result.commit_position}")

        # Tombstone with state expectation
        result = await conn.streams.tombstone(
            stream="temp-data",
            stream_state=StreamState.STREAM_EXISTS,
        )

        # Tombstone at specific revision
        result = await conn.streams.tombstone(
            stream="old-archive",
            revision=5,
        )

asyncio.run(tombstone_stream())
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.