### Install KurrentDB Client with pip

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/getting-started.md

Sets up a virtual environment and installs the kurrentdbclient package version 1.3 using pip.

```bash
python -m venv .venv
source .venv/bin/activate
pip install "kurrentdbclient~=1.3"
```

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

### Install kurrentdbclient using pip

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Install the kurrentdbclient package directly from the Python Package Index using pip.

```bash
pip install kurrentdbclient
```

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

### Install KurrentDB Client with uv

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/getting-started.md

Installs or updates the kurrentdbclient package to version 1.3 using the 'uv' package manager.

```bash
uv add "kurrentdbclient~=1.3"
```

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

### Install KurrentDB with OpenTelemetry Support

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/observability.md

Install the kurrentdbclient package with the opentelemetry option to ensure verified version compatibility.

```bash
pip install kurrentdbclient[opentelemetry]
```

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

### Subscribe to Stream Events from Start

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Use this method to start a catch-up subscription from the first recorded event in a stream. Ensure the stream_name is correctly specified.

```python
# Subscribe from the start of 'stream2'.
with client.subscribe_to_stream(
    stream_name=stream_name2
) as subscription:
    ...
```

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

### Install OpenTelemetry Package Extra

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Install the 'opentelemetry' package extra for KurrentDBClient to ensure verified version compatibility with OpenTelemetry Python packages.

```bash
$ pip install kurrentdbclient[opentelemetry]
```

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

### Import KurrentDBClient class

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Import the KurrentDBClient class from the kurrentdbclient package to start using the client.

```python
from kurrentdbclient import KurrentDBClient
```

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

### Add kurrentdbclient using Poetry

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Add the kurrentdbclient package to your pyproject.toml and install it using Poetry.

```bash
poetry add kurrentdbclient
```

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

### Enable Projection (Sync)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Use the `enable_projection()` method to start a projection that has been disabled.

```python
client.enable_projection(
    name="projection-order-123",
)
```

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

### Enable Projection (Async)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Use the `enable_projection()` method to start a projection that has been disabled.

```python
await client.enable_projection(
    name="projection-order-123",
)
```

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

### Install KurrentDB Client with Poetry

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/getting-started.md

Adds the kurrentdbclient package, specifying a version compatible with 1.3, to a project managed by Poetry.

```bash
poetry add "kurrentdbclient~=1.3"
```

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

### Enable a Projection

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Use `enable_projection()` to start a previously stopped projection. Requires a projection name.

```python
client.enable_projection(projection_name)
```

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

### Run KurrentDB Locally with Docker

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/getting-started.md

Starts a local KurrentDB instance using Docker in insecure mode with specific configurations for projections and HTTP AtomPub.

```bash
docker run --name kurrentdb-node -it -p 2113:2113 \
    docker.kurrent.io/kurrent-lts/kurrentdb:latest \
    --insecure \
    --run-projections=All \
    --enable-atom-pub-over-http
```

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

### Subscribe to All Events from Start

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Initiates a catch-up subscription to retrieve all events from the beginning of the database. This is useful for initial data synchronization or full event history retrieval.

```python
# Subscribe from the first recorded event in the database.
with client.subscribe_to_all() as catchup_subscription:
    ...
```

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

### Add OpenTelemetry with Poetry

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Use Poetry to add the 'opentelemetry' package extra to your pyproject.toml file and install it.

```bash
$ poetry add kurrentdbclient[opentelemetry]
```

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

### Create Persistent Stream Subscription

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Creates a persistent stream subscription from the start of the stream. Ensure the `group_name` and `stream_name` are provided.

```python
# Create a persistent stream subscription from start of the stream.
group_name2 = f"group-{uuid.uuid4()}"
client.create_subscription_to_stream(
    group_name=group_name2,
    stream_name=stream_name2,
)
```

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

### Start Jaeger Locally for OTLP Export

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/observability.md

Run a local Jaeger instance using Docker to receive OTLP telemetry data. Access the UI at http://localhost:16686.

```bash
docker run --name jaeger -d -p 4318:4318 -p 16686:16686 \
  jaegertracing/all-in-one:latest
```

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

### Projection Query Example

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Defines a Javascript projection query that processes events from a specific stream, initializes state, and emits new events.

```javascript
fromStream("order-123")
    .when({
      $init: function(){
        return {
          count: 0,
          list: [null, "2.10", true]
        };
      },
      OrderCreated: function(s,e){
        s.count += 1;
        emit("emitted-order-123", "Emitted", {}, {});
      }
    })
    .outputState()
```

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

### List Continuous Projection Statistics (Sync)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Get a list of statistics for all continuous projections using the synchronous client. The result is a list of ProjectionStatistics objects.

```python
from kurrentdbclient.projections import ProjectionStatistics

statistics = client.list_continuous_projection_statistics()

assert isinstance(statistics, list)
assert 0 < len(statistics)
assert isinstance(statistics[0], ProjectionStatistics)
```

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

### List Continuous Projection Statistics (Async)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Get a list of statistics for all continuous projections using the asynchronous client. The result is a list of ProjectionStatistics objects.

```python
from kurrentdbclient.projections import ProjectionStatistics

statistics = await client.list_continuous_projection_statistics()

assert isinstance(statistics, list)
assert 0 < len(statistics)
assert isinstance(statistics[0], ProjectionStatistics)
```

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

### Initialize KurrentDBClient

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Constructs a KurrentDBClient instance. Connects to an insecure KurrentDB server running locally on port 2113.

```python
import uuid

from kurrentdbclient import KurrentDBClient, NewEvent, StreamState

# Construct KurrentDBClient with a KurrentDB URI. The
# connection string URI specifies that the client should
# connect to an "insecure" server running on port 2113.

client = KurrentDBClient(
    uri="kurrentdb://localhost:2113?Tls=false"
)
```

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

### Demonstrate Async KurrentDB Client Operations

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

This example shows how to use the AsyncKurrentDBClient to connect, append events, read stream events, and subscribe to all events. It requires environment variables KDB_URI and KDB_ROOT_CERTIFICATES for connection.

```python
import asyncio
import os
import uuid

from kurrentdbclient import AsyncKurrentDBClient, NewEvent, StreamState


async def demonstrate_async_client():
    # Construct client.
    client = AsyncKurrentDBClient(
        uri=os.getenv("KDB_URI"),
        root_certificates=os.getenv("KDB_ROOT_CERTIFICATES"),
    )

    # Connect to KurrentDB.
    await client.connect()

    # Append events.
    stream_name = str(uuid.uuid4())
    event1 = NewEvent("OrderCreated", data=b'{}')
    event2 = NewEvent("OrderUpdated", data=b'{}')
    event3 = NewEvent("OrderDeleted", data=b'{}')

    commit_position = await client.append_to_stream(
        stream_name=stream_name,
        current_version=StreamState.NO_STREAM,
        events=[event1, event2, event3]
    )

    # Get stream events.
    recorded = await client.get_stream(stream_name)
    assert len(recorded) == 3
    assert recorded[0] == event1
    assert recorded[1] == event2
    assert recorded[2] == event3

    # Subscribe to all events.
    received = []
    async with await client.subscribe_to_all(commit_position=0) as subscription:
        async for event in subscription:
            received.append(event)
            if event.commit_position == commit_position:
                break
    assert received[-3] == event1
    assert received[-2] == event2
    assert received[-1] == event3

    # Close the client.
    await client.close()


# Run the demo.
asyncio.run(
    demonstrate_async_client()
)
```

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

### subscribe_to_index

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Starts a catch-up subscription using a secondary index to retrieve events in the order they were recorded. Allows specifying a starting commit position.

```APIDOC
## subscribe_to_index

### Description
Starts a catch-up subscription using a secondary index to retrieve events in the order they were recorded. Allows specifying a starting commit position.

### Method
client.subscribe_to_index()

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
* **index_name** (string) - Required - The name of the index to read from (e.g., "et-EventType").
* **commit_position** (int) - Optional - A commit position from which to start reading. Must be an existing commit position. Defaults to None.
* **timeout** (float) - Optional - Maximum duration in seconds for the gRPC operation.
* **credentials** (any) - Optional - Credentials to override connection string credentials.

### Response
Returns a "catch-up subscription" iterator.
```

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

### Consumer Span Example

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/observability.md

An example of an instrumentor span generated during a catch-up subscription operation. It includes attributes detailing the database operation, system, user, event, stream, and server connection.

```json
{
    "name": "streams.subscribe",
    "context": {
        "trace_id": "0x5ad5e1bcff7f33cb44b93d470bd34554",
        "span_id": "0x446cf48b1bb9e574",
        "trace_state": "[]"
    },
    "kind": "SpanKind.CONSUMER",
    "parent_id": "0x1496f8ba3507977b",
    "start_time": "2026-02-17T14:16:20.810515Z",
    "end_time": "2026-02-17T14:16:20.810605Z",
    "status": {
        "status_code": "OK"
    },
    "attributes": {
        "db.operation": "streams.subscribe",
        "db.system": "kurrentdb",
        "db.user": "admin",
        "db.kurrentdb.event.id": "4ca26d3e-cbec-477e-9e59-d9248d8a3aef",
        "db.kurrentdb.event.type": "UserRegistered",
        "db.kurrentdb.stream": "user-123",
        "db.kurrentdb.subscription.id": "5da1a8c8-3dec-441e-8b6f-7514c797b1b4",
        "server.address": "localhost",
        "server.port": "2113"
    },
    "events": [],
    "links": [],
    "resource": {
        "attributes": {
            "telemetry.sdk.language": "python",
            "telemetry.sdk.name": "opentelemetry",
            "telemetry.sdk.version": "1.39.1",
            "service.name": "kurrentdb"
        },
        "schema_url": ""
    }
}
```

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

### Construct KurrentDBClient with Environment Variables

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Instantiate the KurrentDBClient using environment variables for the connection URI and root certificates. This is useful for configuring the client without hardcoding sensitive information.

```python
import os

client = KurrentDBClient(
    uri=os.getenv("KDB_URI"),
    root_certificates=os.getenv("KDB_ROOT_CERTIFICATES"),
)
```

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

### Get Stream Events Backwards (Async)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/reading-events.md

Retrieves all events from a specified stream in reverse chronological order using an asynchronous client. The loop breaks after the first event is processed, effectively getting the last event.

```python
# Get all events backwards from the end
for event in await client.get_stream(
    stream_name="order-123",
    backwards=True,
):
    assert event.stream_position == 2
    break
```

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

### Connect to KurrentDB with Sync Client

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/getting-started.md

Instantiates the synchronous KurrentDB client using a predefined connection string.

```python
from kurrentdbclient import KurrentDBClient

client = KurrentDBClient(connection_string)
```

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

### Get Stream Events Backwards (Sync)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/reading-events.md

Retrieves all events from a specified stream in reverse chronological order using a synchronous client. The loop breaks after the first event is processed, effectively getting the last event.

```python
# Get all events backwards from the end
for event in client.get_stream(
    stream_name="order-123",
    backwards=True,
):
    assert event.stream_position == 2
    break
```

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

### enable_projection

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Enables a disabled projection, allowing it to start running.

```APIDOC
## Enable Projection

Use the `enable_projection()` method to start a projection that has been disabled.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **name** (string) - Required - Name of the projection.
- **timeout** (integer) - Optional - Maximum duration of operation (in seconds). Defaults to `None`.
- **credentials** (object) - Optional - Override credentials derived from client configuration. Defaults to `None`.

### Request Example
```json
{
  "name": "projection-order-123"
}
```

### Response
#### Success Response (200)
Returns `None` on success.

#### Response Example
```json
null
```
```

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

### Enable Projection

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Enables (starts running) a previously disabled projection. Requires leader.

```APIDOC
## Enable Projection

### Description
Enables (starts running) a previously disabled projection. Requires leader.

### Method
`enable_projection(name: str, timeout: float = None, credentials: Any = None)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
* **name** (str) - Required - The name of the projection to be enabled.
* **timeout** (float) - Optional - Maximum duration in seconds for the gRPC operation.
* **credentials** (Any) - Optional - Override call credentials derived from the connection string URI.

### Request Example
```python
client.enable_projection(projection_name)
```

### Response
#### Success Response
(No specific success response details provided in source)

#### Response Example
(No specific response example provided in source)
```

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

### Connect to KurrentDB with Async Client

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/getting-started.md

Instantiates the asynchronous KurrentDB client using a predefined connection string.

```python
from kurrentdbclient import AsyncKurrentDBClient

client = AsyncKurrentDBClient(connection_string)
```

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

### Register a new dog

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Registers a new dog with the system and retrieves its initial state. Asserts the initial properties of the newly registered dog.

```python
dog_id = register_dog('Fido')

dog = get_dog(dog_id)
assert dog.name == 'Fido'
assert dog.tricks == []
assert dog.version == 0
assert dog.is_from_snapshot is False
```

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

### Get Projection Statistics

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Retrieves statistics for a specific projection. This method requires leader access.

```python
statistics = client.get_projection_statistics(projection_name)
```

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

### Initialize Tracer Provider for OTLP Export

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Configure OpenTelemetry to export trace data to an OTLP-compatible collector like Jaeger. Requires 'opentelemetry-exporter-otlp-proto-http' package.

```python
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import set_tracer_provider

resource = Resource.create(
    attributes={
        SERVICE_NAME: "kurrentdb",
    }
)
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")))
set_tracer_provider(provider)
```

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

### CreateReq Options with Engine Version

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/prds/projection-engine-v2.md

Defines the `CreateReq.Options` message including the new `engine_version` field for selecting the projection engine. V1 is the default (0 or 1), while V2 is explicitly set to 2.

```proto
message CreateReq {
  Options options = 1;
  message Options {
    oneof mode {
      event_store.client.Empty one_time = 1;
      Transient transient = 2 [deprecated = true];
      Continuous continuous = 3;
    }
    string query = 4;
    int32  engine_version = 5;   // NEW. 0 or 1 = v1 (default), 2 = v2
    // ...
  }
}
```

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

### Read Index

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/reading-events.md

Reads events from a secondary index in KurrentDB, optionally starting from a specific commit position and with a limit.

```APIDOC
## read_index

### Description
Reads events from a secondary index in KurrentDB.

### Method
GET (or equivalent for async)

### Endpoint
/index/{index_name}

### Parameters
#### Path Parameters
- **index_name** (string) - Required - Name of the secondary index (e.g., "et-OrderCreated"). The "$idx-" prefix is optional.

#### Query Parameters
- **commit_position** (integer) - Optional - Position from which to start reading events.
- **limit** (integer) - Optional - Maximum number of events to return.
- **timeout** (integer) - Optional - Maximum duration of the operation in seconds.
- **credentials** (object) - Optional - Override credentials.

### Response
#### Success Response (200)
An iterable of RecordedEvent objects.

#### Response Example
```json
{
  "example": "Iterable of RecordedEvent objects"
}
```
```

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

### Connect to KurrentDB and Append Events (Async)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/subscriptions.md

Connects to KurrentDB using the asynchronous client and appends the first event to a stream. Requires setting up the client and constructing NewEvent objects.

```python
from kurrentdbclient import AsyncKurrentDBClient, NewEvent, StreamState

# Connect to KurrentDB
uri = "kurrentdb://127.0.0.1:2113?tls=false"
client = AsyncKurrentDBClient(uri)

# Construct new event objects
event1 = NewEvent(
    type="OrderCreated",
    data=b'{"order_id": "order-123"}',
)
event2 = NewEvent(
    type="OrderUpdated",
    data=b'{"status": "processing"}',
)
event3 = NewEvent(
    type="OrderUpdated",
    data=b'{"status": "shipped"}',
)

# Append the first event to a new stream
commit_position = await client.append_to_stream(
    stream_name="order-123",
    current_version=StreamState.NO_STREAM,
    events=[event1],
)
```

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

### Get Subscription Info (All Streams)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Retrieves information for a persistent subscription created for all streams. Requires the group name.

```python
subscription_info = client.get_subscription_info(
    group_name=group_name1,
)
```

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

### Start Catch-up Subscription

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Initiates a catch-up subscription to receive all recorded events in the database. This method returns an iterator that blocks until new events are recorded, supporting exactly-once processing semantics.

```python
# Start a catch-up subscription from last recorded position.
# This method returns a "catch-up subscription" object,
# which can be iterated over to obtain recorded events.
# The iterator will not stop when there are no more recorded
# events to be returned, but instead will block, and then continue
# when further events are recorded. It can be used as a context
# manager so that the underlying streaming gRPC call to the database
```

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

### Get Stream Metadata

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Fetches the metadata and version for a specific stream. Use this to inspect or retrieve stream-specific configuration.

```python
metadata, metadata_version = client.get_stream_metadata(stream_name=stream_name1)
```

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

### Get Projection Statistics

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Retrieves statistics for a specific projection by its name. This method can be used in both synchronous and asynchronous contexts.

```APIDOC
## Get Projection Statistics

Use the `get_projection_statistics()` method to get statistics for a projection.

### Parameters
#### Query Parameters
- **name** (string) - Required - Name of the projection.
- **timeout** (float) - Optional - Maximum duration of operation (in seconds).
- **credentials** (object) - Optional - Override credentials derived from client configuration.

### Response
#### Success Response
Returns a [`ProjectionStatistics`](#the-projectionstatistics-class) object.

### Request Example
```python
statistics = client.get_projection_statistics(name="projection-order-123")
```

### Response Example
```json
{
  "status": "Running"
}
```
```

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

### Synchronous Event Processing with Checkpointing

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/subscriptions.md

Demonstrates how to process events from a subscription and save checkpoints synchronously. It retrieves the last checkpoint, subscribes, and updates state or saves checkpoints based on the item type. Requires `kurrentdbclient` and a `Projection` class.

```python
from kurrentdbclient import Checkpoint, RecordedEvent

def process_events_with_checkpointing(client, projection):
    # Get the last checkpoint
    last_commit_position = projection.get_last_checkpoint()

    # Subscribe using the last checkpoint
    with client.subscribe_to_all(
        commit_position=last_commit_position,
        include_checkpoints=True
    ) as subscription:
        for item in subscription:
            if type(item) is RecordedEvent:
                # Regular event processing
                new_state = {"key": "value"}
                # Record commit position with new state
                projection.update_state(new_state, item.commit_position)

            elif type(item) is Checkpoint:
                # Record commit position
                projection.save_checkpoint(item.commit_position)

            break  # <-- so we can continue with the examples



class Projection:
    def __init__(self):
        self._last_checkpoint = None
        self._current_state = {}

    def get_last_checkpoint(self):
        return self._last_checkpoint

    def save_checkpoint(self, checkpoint):
        # Only update if checkpoint is greater than last recorded.
        if (
            self._last_checkpoint is None
            or checkpoint > self._last_checkpoint
        ):
            self._last_checkpoint = checkpoint

    def update_state(self, state, checkpoint):
        # Only update if checkpoint is greater than last recorded.
        if (
            self._last_checkpoint is not None
            and checkpoint <= self._last_checkpoint
        ):
            msg = f"Checkpoint conflict: {checkpoint} <= {self._last_checkpoint}"
            raise ValueError(msg)
        self._last_checkpoint = checkpoint
        self._current_state = state


process_events_with_checkpointing(client, Projection())

```

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

### Connect to KurrentDB and Append Events (Sync)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/subscriptions.md

Connects to KurrentDB using the synchronous client and appends multiple events to a stream. Requires setting up the client and constructing NewEvent objects.

```python
from kurrentdbclient import KurrentDBClient, NewEvent, StreamState

# Connect to KurrentDB
uri = "kurrentdb://127.0.0.1:2113?tls=false"
client = KurrentDBClient(uri)

# Construct new event objects
event1 = NewEvent(
    type="OrderCreated",
    data=b'{"order_id": "order-123"}',
)
event2 = NewEvent(
    type="OrderUpdated",
    data=b'{"status": "processing"}',
)
event3 = NewEvent(
    type="OrderUpdated",
    data=b'{"status": "shipped"}',
)

# Append the first event to a new stream
commit_position = client.append_to_stream(
    stream_name="order-123",
    current_version=StreamState.NO_STREAM,
    events=[event1],
)

# Append second and third event to the same stream.
client.append_to_stream(
    stream_name="order-123",
    current_version=0,
    events=[event2, event3],
)
```

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

### Read Stream Events Backwards

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Retrieves all recorded events from a stream in reverse order, starting from the last recorded event.

```python
events = client.get_stream(
    stream_name=stream_name1,
    backwards=True,
)

assert len(events) == 3
assert events[0] == event3
assert events[1] == event2
assert events[2] == event1
```

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

### Create Subscription to All Streams

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/persistent-subscriptions.md

Use `create_subscription_to_all` to create a persistent subscription for consuming the global transaction log. Various filtering and strategy options are available.

```python
await client.create_subscription_to_all(
    group_name="all-streams-subscription",
    from_end=True,
    filter_exclude=["System-*"],
    filter_by_stream_name=True,
    consumer_strategy="RoundRobin"
)
```

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

### Handle Stream Not Found Error

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/reading-events.md

Reading a stream that does not exist will raise a `NotFoundError`. This example demonstrates how to catch this specific exception.

```python
from kurrentdbclient.exceptions import NotFoundError

try:
    with client.read_stream(
        stream_name="not-a-stream"
    ) as events:
        ...
except NotFoundError:
    print("Success: Stream does not exist")
except Exception as e:
    print(f"Shouldn't get here")
```

```python
from kurrentdbclient.exceptions import NotFoundError

try:
    async with await client.read_stream(
        stream_name="not-a-stream"
    ) as events:
        ...
except NotFoundError:
    print("Success: Stream does not exist")
except Exception as e:
    print(f"Shouldn't get here")
```

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

### Get Current Version of a Non-Existent Stream

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Retrieves the current version for a stream that does not exist. The method returns StreamState.NO_STREAM in this case.

```python
current_version = client.get_current_version(
    stream_name='does-not-exist'
)

assert current_version is StreamState.NO_STREAM
```

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

### List All Subscriptions (Sync)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/persistent-subscriptions.md

Returns a list of SubscriptionInfo objects for all existing persistent subscriptions using the synchronous client.

```python
client.list_subscriptions()
```

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

### Get Subscription Info

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/persistent-subscriptions.md

Retrieves information about a specific persistent subscription group. This can be for a particular stream or for the global transaction log.

```APIDOC
## Get Subscription Info

### Description
Use `get_subscription_info()` to get a [`SubscriptionInfo`](#subscription-info) object for a persistent subscription group.

### Method
`get_subscription_info(group_name: str, stream_name: Optional[str] = None, timeout: Optional[float] = None, credentials: Optional[Any] = None)`

### Parameters
#### Path Parameters
- `group_name` (string) - Required - Name of persistent subscription group.
- `stream_name` (string) - Optional - Name of stream.
- `timeout` (float) - Optional - Maximum duration of operation (in seconds).
- `credentials` (Any) - Optional - Override credentials derived from client configuration.

### Request Example
```python
# For a specific stream
client.get_subscription_info(
    group_name="stream-subscription",
    stream_name="order-123",
)

# For the global transaction log
client.get_subscription_info(
    group_name="transaction-log-subscription",
)
```

### Response
#### Success Response (200)
- `SubscriptionInfo` object describing the subscription.
```

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

### KurrentDB Command Functions

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Defines functions to register a new dog, record a learned trick, and snapshot a dog's state. Note the different stream naming conventions for regular events and snapshots.

```python
def register_dog(name):
    dog_id = str(uuid.uuid4())
    event = NewEvent(
        type='DogRegistered',
        data=serialize({'name': name}),
    )
    client.append_to_stream(
        stream_name=dog_id,
        current_version=StreamState.NO_STREAM,
        events=event,
    )
    return dog_id

def record_trick_learned(dog_id, trick):
    dog = get_dog(dog_id)
    event = NewEvent(
        type='DogLearnedTrick',
        data=serialize({'trick': trick}),
    )
    client.append_to_stream(
        stream_name=dog_id,
        current_version=dog.version,
        events=event,
    )

def snapshot_dog(dog_id):
    dog = get_dog(dog_id)
    event = NewEvent(
        type='Snapshot',
        data=serialize({'name': dog.name, 'tricks': dog.tricks}),
        metadata=serialize({'version': dog.version}),
    )
    client.append_to_stream(
        stream_name=make_snapshot_stream_name(dog_id),
        current_version=StreamState.ANY,
        events=event,
    )
```

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

### Configure OTLP Exporter for Telemetry Data

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/observability.md

Initialize the global tracer provider with an OTLPSpanExporter to send telemetry data to an OpenTelemetry collector. Requires opentelemetry-exporter-otlp-proto-http.

```python
from opentelemetry.exporter.otlp.proto.http.trace_exporter import \
    OTLPSpanExporter
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import set_tracer_provider

resource = Resource.create(
    attributes={
        SERVICE_NAME: "kurrentdb",
    }
)
provider = TracerProvider(resource=resource)
provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
    )
)
set_tracer_provider(provider)
```

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

### Get Projection State

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Retrieves the current state of a specified projection. The state is returned as a ProjectionState object with a 'value' attribute.

```APIDOC
## Get Projection State

Use the `get_projection_state()` method to get a projection's current state.

### Parameters
#### Path Parameters
- `name` (string) - Required - Name of the projection.
- `partition` (string) - Optional - Projection partition.
#### Query Parameters
- `timeout` (float) - Optional - Maximum duration of operation (in seconds).
- `credentials` (object) - Optional - Override credentials derived from client configuration.

### Response
#### Success Response (200)
- `value` (object) - The current state of the projection.

### Request Example
```python
state = client.get_projection_state("projection-order-123")
```

### Response Example
```json
{
  "value": {
    "count": 1,
    "list": [
      null,
      "2.10",
      true
    ]
  }
}
```
```

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

### Subscribe to Stream Events from Specific Position

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Use this method to start a catch-up subscription from a particular stream position. The stream_position is 0-indexed.

```python
# Subscribe to stream2, from the second recorded event.
with client.subscribe_to_stream(
    stream_name=stream_name2,
    stream_position=1,
) as subscription:
    ...
```

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

### Sync Subscription Retry Logic

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/subscriptions.md

Implement retry logic for synchronous subscriptions to handle dropped connections or consumer slowness. This example retries up to 5 times with a 5-second delay.

```python
import time
from kurrentdbclient.exceptions import ConsumerTooSlowError, GrpcError

projection = Projection()
retries = 5

while True:
    try:
        process_events_with_checkpointing(client, projection)
        break
    except (ConsumerTooSlowError, GrpcError):
        if retries <= 0:
            raise
        retries -= 1
        time.sleep(5)
        continue
```

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

### Read Stream Events from a Specific Position

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Retrieves events from a stream starting from a specified position up to the end. Stream positions are zero-based.

```python
events = client.get_stream(
    stream_name=stream_name1,
    stream_position=1,
)

assert len(events) == 2
assert events[0] == event2
assert events[1] == event3
```

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

### Asynchronous Event Processing with Checkpointing

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/subscriptions.md

Demonstrates how to process events from a subscription and save checkpoints asynchronously. It retrieves the last checkpoint, subscribes, and updates state or saves checkpoints based on the item type. Requires `kurrentdbclient` and an async `Projection` class.

```python
from kurrentdbclient import Checkpoint, RecordedEvent

async def process_events_with_checkpointing(client, projection):
    # Get the last checkpoint
    last_commit_position = await projection.get_last_checkpoint()

    # Subscribe using the last checkpoint
    async with await client.subscribe_to_all(
        commit_position=last_commit_position,
        include_checkpoints=True
    ) as subscription:

        async for item in subscription:
            if type(item) is RecordedEvent:
                # Regular event processing
                new_state = {"key": "value"}
                # Record commit position with new state
                await projection.update_state(new_state, item.commit_position)

            elif type(item) is Checkpoint:
                # Record commit position
                await projection.save_checkpoint(item.commit_position)

            break  # <-- so we can continue with the examples


class Projection:
    def __init__(self):
        self._last_checkpoint = None
        self._current_state = {}

    async def get_last_checkpoint(self):
        return self._last_checkpoint

    async def save_checkpoint(self, checkpoint):
        # Only update if checkpoint is greater than last recorded.
        if (
            self._last_checkpoint is None
            or checkpoint > self._last_checkpoint
        ):
            self._last_checkpoint = checkpoint

    async def update_state(self, state, checkpoint):
        # Only update if checkpoint is greater than last recorded.
        if (
            self._last_checkpoint is not None
            and checkpoint <= self._last_checkpoint
        ):
            msg = f"Checkpoint conflict: {checkpoint} <= {self._last_checkpoint}"
            raise ValueError(msg)
        self._last_checkpoint = checkpoint
        self._current_state = state


await process_events_with_checkpointing(client, Projection())

```

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

### Read All Events

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/reading-events.md

Reads all events from the KurrentDB log starting from a specific commit position. An InvalidCommitPositionError is raised if the commit position does not exist.

```APIDOC
## read_all

### Description
Reads all events from the KurrentDB log.

### Method
GET (or equivalent for async)

### Endpoint
/events

### Parameters
#### Query Parameters
- **commit_position** (integer) - Required - The position from which to start reading events.
- **resolve_links** (boolean) - Optional - If True, resolves link events and returns the linked events.
- **filter_include** (list of strings) - Optional - A list of patterns to include for filtering events. By default, filters by event type. Can filter by stream name if `filter_by_stream_name` is True.
- **filter_exclude** (list of strings) - Optional - A list of patterns to exclude for filtering events. By default, filters by event type. Can filter by stream name if `filter_by_stream_name` is True.
- **filter_by_stream_name** (boolean) - Optional - If True, filters by stream name instead of event type.

### Response
#### Success Response (200)
An iterable of RecordedEvent objects.

#### Response Example
```json
{
  "example": "Iterable of RecordedEvent objects"
}
```
```

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

### Handle Stream Not Found Error (Sync)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/subscriptions.md

Demonstrates catching a NotFoundError when subscribing to a non-existent stream using a synchronous client. Includes necessary import.

```python
from kurrentdbclient.exceptions import NotFoundError

try:
    with client.subscribe_to_stream(
        stream_name="not-a-stream"
    ) as subscription:
        ...
except NotFoundError:
    print("Success: Stream does not exist")
except Exception as e:
    print(f"Shouldn't get here")
```

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

### Get Projection Statistics (Async)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Retrieve statistics for a specific projection using the asynchronous client. Ensure the projection name is provided.

```python
statistics = await client.get_projection_statistics(
    name="projection-order-123",
)
assert "Running" == statistics.status
```

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

### List All Subscriptions (Async)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/persistent-subscriptions.md

Returns a list of SubscriptionInfo objects for all existing persistent subscriptions using the asynchronous client.

```python
await client.list_subscriptions()
```

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

### Get Projection Statistics (Sync)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Retrieve statistics for a specific projection using the synchronous client. Ensure the projection name is provided.

```python
statistics = client.get_projection_statistics(
    name="projection-order-123",
)
assert "Running" == statistics.status
```

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

### Read Subscription to All

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/persistent-subscriptions.md

Starts consuming events from the global transaction log using the `read_subscription_to_all()` method. This allows for continuous event processing and acknowledgment.

```APIDOC
## read_subscription_to_all()

### Description
Use `read_subscription_to_all()` to start consuming events from the global transaction log. This method returns a `PersistentSubscription` object for processing events.

### Method
```
client.read_subscription_to_all(
    group_name: str,
    event_buffer_size: int = 150,
    max_ack_batch_size: int = 50,
    max_ack_delay: float = 0.2,
    stopping_grace: float = 0.2,
    timeout: Optional[float] = None,
    credentials: Optional[Any] = None
)
```

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
* **group_name** (str) - Required - Name of persistent subscription group.
* **event_buffer_size** (int) - Optional - Number of events in consumer buffer. Defaults to `150`.
* **max_ack_batch_size** (int) - Optional - Number of acknowledgements before sending all to server. Defaults to `50`.
* **max_ack_delay** (float) - Optional - Amount of time (in seconds) before sending acknowledgements to server. Defaults to `0.2`.
* **stopping_grace** (float) - Optional - Amount of time (in seconds) to allow server to receive acknowledgements when consumer is stopping. Defaults to `0.2`.
* **timeout** (Optional[float]) - Optional - Maximum duration of operation (in seconds). Defaults to `None`.
* **credentials** (Optional[Any]) - Optional - Override credentials derived from client configuration. Defaults to `None`.

### Returns
A [`PersistentSubscription`](#consuming-events) object.

### Example
#### sync
```python
# Connect to a persistent subscription for all events
with client.read_subscription_to_all(
    group_name="transaction-log-subscription"
) as subscription:

    # Process events and acknowledge them
    for event in subscription:
        try:
            # Process the event
            print(f"Processing event: {event.type} from stream {event.stream_name}")

            # Acknowledge successful processing
            subscription.ack(event)

        except Exception as e:
            # Handle processing errors
            print(f"Error processing event: {e}")
            subscription.nack(event, action="retry")

        break  # <- so we can continue with the examples
```

#### async
```python
# Connect to a persistent subscription for all events
async with await client.read_subscription_to_all(
    group_name="transaction-log-subscription"
) as subscription:

    # Process events and acknowledge them
    async for event in subscription:
        try:
            # Process the event
            print(f"Processing event: {event.type} from stream {event.stream_name}")

            # Acknowledge successful processing
            await subscription.ack(event)

        except Exception as e:
            # Handle processing errors
            print(f"Error processing event: {e}")
            await subscription.nack(event, action="retry")

        break  # <- so we can continue with the examples
```
```

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

### Create Persistent Subscription (Async)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/persistent-subscriptions.md

This snippet demonstrates how to initiate an asynchronous persistent subscription to a stream. It requires an AsyncKurrentDBClient instance.

```python
from kurrentdbclient import AsyncKurrentDBClient

# Connect to KurrentDB
connection_string = "kurrentdb://127.0.0.1:2113?tls=false"
client = AsyncKurrentDBClient(connection_string)


```

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

### Basic Subscription to All Events (Sync)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/subscriptions.md

Subscribes to all events in the global transaction log using a synchronous client. Use this to iterate through all recorded events.

```python
# Subscribe to all events in global transaction log
with client.subscribe_to_all() as subscription:
    # Iterate through the subscription with a 'for' loop
    for event in subscription:
        print(f"Event: {event.type} at position {event.commit_position}")
        break  # <-- so we can continue with the examples
```

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

### Read Subscription to Stream

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/persistent-subscriptions.md

Starts consuming events from a persistent subscription to a specific stream. Allows for detailed configuration of buffering, acknowledgements, and timeouts.

```APIDOC
## read_subscription_to_stream

### Description
Starts consuming events from a persistent subscription to a specific stream. This method returns a `PersistentSubscription` object that can be iterated over to process events.

### Method
`client.read_subscription_to_stream(group_name: str, stream_name: str, event_buffer_size: int = 150, max_ack_batch_size: int = 50, max_ack_delay: float = 0.2, stopping_grace: float = 0.2, timeout: float = None, credentials: object = None)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
* **`group_name`** (string) - Required - Name of persistent subscription group.
* **`stream_name`** (string) - Required - Name of stream from which to consume events.
* **`event_buffer_size`** (integer) - Optional - Number of events in consumer buffer. Defaults to `150`.
* **`max_ack_batch_size`** (integer) - Optional - Number of acknowledgements before sending all to server. Defaults to `50`.
* **`max_ack_delay`** (float) - Optional - Amount of time (in seconds) before sending acknowledgements to server. Defaults to `0.2`.
* **`stopping_grace`** (float) - Optional - Amount of time (in seconds) to allow server to receive acknowledgements when consumer is stopping. Defaults to `0.2`.
* **`timeout`** (float) - Optional - Maximum duration of operation (in seconds). Defaults to `None`.
* **`credentials`** (object) - Optional - Override credentials derived from client configuration.

### Request Example
```python
with client.read_subscription_to_stream(
    group_name="stream-subscription",
    stream_name="order-123"
) as subscription:
    for event in subscription:
        # Process event and acknowledge
        subscription.ack(event)
```

### Response
#### Success Response (200)
Returns a [`PersistentSubscription`](#consuming-events) object.

#### Response Example
None explicitly defined, but implies an iterable subscription object.
```

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

### Filter Events by Type (Exclude)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/reading-events.md

Reads events where the event type does not start with 'Order'. Server-side filtering by event type is the default behavior.

```python
with client.read_all(
    filter_exclude=["Order.*"]
) as events:
    ...
```

```python
async with await client.read_all(
    filter_exclude=["Order.*"]
) as events:
    ...
```

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

### Create Synchronous Projection

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/projections.md

Creates a projection using the synchronous client. Ensure `emit_enabled` is True if the query uses `.emit()`.

```python
client.create_projection(
    name="projection-order-123",
    query=projection_query,
    emit_enabled=True,
    track_emitted_streams=True,
)
```

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

### Filter Events by Type (Include)

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/docs/api/reading-events.md

Reads events where the event type starts with 'Order'. Server-side filtering by event type is the default behavior.

```python
with client.read_all(
    filter_include=["Order.*"]
) as events:
    ...
```

```python
async with await client.read_all(
    filter_include=["Order.*"]
) as events:
    ...
```

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

### Subscribe to Events After a Specific Commit Position

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

This example demonstrates restarting a subscription to capture events that occurred after a specific commit position. It appends new events while the subscription is active and then waits for them.

```python
# Append another event.
event6 = NewEvent(type='OrderDeleted', data=b'{}')
client.append_to_stream(
    stream_name=stream_name2,
    current_version=1,
    events=[event6],
)

# Restart subscribing to all events after the
# commit position of the last received event.
subscription_to_all = SubscribeToAll(
    client=client,
    commit_position=subscription_to_all.last_commit_position
)

# Wait for event6.
subscription_to_all.wait_for_event(event6)

# Append three more events to a new stream.
stream_name3 = str(uuid.uuid4())
event7 = NewEvent(type='OrderCreated', data=b'{}')
event8 = NewEvent(type='OrderUpdated', data=b'{}')
event9 = NewEvent(type='OrderDeleted', data=b'{}')

client.append_to_stream(
    stream_name=stream_name3,
    current_version=StreamState.NO_STREAM,
    events=[event7, event8, event9],
)

# Wait for events 7, 8 and 9.
subscription_to_all.wait_for_event(event7)
subscription_to_all.wait_for_event(event8)
subscription_to_all.wait_for_event(event9)

# Stop the subscription thread.
subscription_to_all.stop()
```

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

### Get Last Commit Position

Source: https://github.com/pyeventsourcing/kurrentdbclient/blob/1.3/README.md

Retrieves the commit position of the last recorded event in the database. This is useful for tracking the latest state or progress.

```python
commit_position = client.get_commit_position()
```