kurrentdbclient 1.2.2
Python gRPC Client for KurrentDB
License: BSD-3-Clause
Requires Python: >=3.10

<a href="https://kurrent.io">
  <picture>
    <source media="(prefers-color-scheme: dark)" srcset="https://github.com/pyeventsourcing/kurrentdbclient/raw/1.0/KurrentLogo-White.png.png">
    <source media="(prefers-color-scheme: light)" srcset="https://github.com/pyeventsourcing/kurrentdbclient/raw/1.0/KurrentLogo-Black.png">
    <img alt="Kurrent" src="https://github.com/pyeventsourcing/kurrentdbclient/raw/1.0/KurrentLogo-Plum.png" height="50%" width="50%">
  </picture>
</a>

Please note: following the rebranding of EventStoreDB to KurrentDB, this package
is a rebranding of the [`esdbclient`](https://pypi.org/project/esdbclient) package.

# Python gRPC Client for KurrentDB

This [Python package](https://pypi.org/project/kurrentdbclient/) provides multithreaded and asyncio Python
clients for the [KurrentDB](https://kurrent.io/) database.

The multithreaded `KurrentDBClient` is described in detail below. Please scroll
down for <a href="#asyncio-client">information</a> about `AsyncKurrentDBClient`.

These clients have been developed and are being maintained in a collaboration
with the KurrentDB team, and are officially supported by Kurrent Inc.
Although not all aspects of the KurrentDB gRPC API are implemented, most
features are presented in an easy-to-use interface.

These clients have been tested to work with KurrentDB version 25.1 and 25.0, EventStoreDB
versions 24.10 and EventStoreDB 23.10, both with and without SSL/TLS, in both single-server and
cluster modes, across Python versions 3.10, 3.11, 3.12, 3.13, and 3.14.

The test suite has 100% line and branch coverage. The code has typing annotations
checked strictly with mypy. The code is formatted with black and isort, and checked
with flake8. Poetry is used for package management during development, and for
building and publishing distributions to [PyPI](https://pypi.org/project/kurrentdbclient/).

For an example of usage, see the [eventsourcing-kurrentdb](
https://github.com/pyeventsourcing/eventsourcing-kurrentdb) package.


<!-- TOC -->
* [Synopsis](#synopsis)
* [Install package](#install-package)
  * [From PyPI](#from-pypi)
  * [With Poetry](#with-poetry)
* [KurrentDB server](#kurrentdb-server)
  * [Run container](#run-container)
  * [Stop container](#stop-container)
* [KurrentDB client](#kurrentdb-client)
  * [Import class](#import-class)
  * [Construct client](#construct-client)
* [Connection strings](#connection-strings)
  * [Two schemes](#two-schemes)
  * [User info string](#user-info-string)
  * [Query string](#query-string)
  * [Examples](#examples)
* [Event objects](#event-objects)
  * [New events](#new-events)
  * [Recorded events](#recorded-events)
* [Streams](#streams)
  * [Append events](#append-events)
  * [Idempotent append operations](#idempotent-append-operations)
  * [Multi-stream append](#multi-stream-append)
  * [Read stream events](#read-stream-events)
  * [Get current version](#get-current-version)
  * [How to implement snapshotting with KurrentDB](#how-to-implement-snapshotting-with-kurrentdb)
  * [Read all events](#read-all-events)
  * [Get commit position](#get-commit-position)
  * [Get stream metadata](#get-stream-metadata)
  * [Set stream metadata](#set-stream-metadata)
  * [Delete stream](#delete-stream)
  * [Tombstone stream](#tombstone-stream)
  * [Read using secondary index](#read-index)
* [Catch-up subscriptions](#catch-up-subscriptions)
  * [Subscribe to all events](#subscribe-to-all-events)
  * [Subscribe to stream events](#subscribe-to-stream-events)
  * [Subscribe using secondary index](#subscribe-to-index)
  * [How to implement exactly-once event processing](#how-to-implement-exactly-once-event-processing)
* [Persistent subscriptions](#persistent-subscriptions)
  * [Create subscription to all](#create-subscription-to-all)
  * [Read subscription to all](#read-subscription-to-all)
  * [How to write a persistent subscription consumer](#how-to-write-a-persistent-subscription-consumer)
  * [Update subscription to all](#update-subscription-to-all)
  * [Create subscription to stream](#create-subscription-to-stream)
  * [Read subscription to stream](#read-subscription-to-stream)
  * [Update subscription to stream](#update-subscription-to-stream)
  * [Replay parked events](#replay-parked-events)
  * [Get subscription info](#get-subscription-info)
  * [List subscriptions](#list-subscriptions)
  * [List subscriptions to stream](#list-subscriptions-to-stream)
  * [Delete subscription](#delete-subscription)
* [Projections](#projections)
  * [Create projection](#create-projection)
  * [Get projection state](#get-projection-state)
  * [Update projection](#update-projection)
  * [Get projection statistics](#get-projection-statistics)
  * [List all projection statistics](#list-all-projection-statistics)
  * [List continuous projection statistics](#list-continuous-projection-statistics)
  * [Disable projection](#disable-projection)
  * [Enable projection](#enable-projection)
  * [Abort projection](#abort-projection)
  * [Reset projection](#reset-projection)
  * [Delete projection](#delete-projection)
  * [Restart projections subsystem](#restart-projections-subsystem)
* [Call credentials](#call-credentials)
  * [Construct call credentials](#construct-call-credentials)
* [Connection](#connection)
  * [Reconnect](#reconnect)
  * [Close](#close)
* [Asyncio client](#asyncio-client)
  * [Synopsis](#synopsis-1)
  * [FastAPI](#fastapi)
* [Notes](#notes)
  * [Regular expression filters](#regular-expression-filters)
  * [Reconnect and retry method decorators](#reconnect-and-retry-method-decorators)
* [Instrumentation](#instrumentation)
  * [OpenTelemetry](#open-telemetry)
* [Communities](#communities)
* [Contributors](#contributors)
  * [Install Poetry](#install-poetry)
  * [Setup for PyCharm users](#setup-for-pycharm-users)
  * [Setup from command line](#setup-from-command-line)
  * [Project Makefile commands](#project-makefile-commands)
<!-- TOC -->

## Synopsis<a id="synopsis"></a>

The `KurrentDBClient` class can be imported from the `kurrentdbclient` package.

Probably the three most useful methods of `KurrentDBClient` are:

* `append_to_stream()` This method can be used to record new events in a particular
"stream". This is useful, for example, when executing a command in an application
that mutates an aggregate. This method is "atomic" in that either all or none of
the events will be recorded.

* `get_stream()` This method can be used to retrieve all the recorded
events in a "stream". This is useful, for example, when reconstructing
an aggregate from recorded events before executing a command in an
application that creates new events.

* `subscribe_to_all()` This method can be used to receive all recorded events in
the database. This is useful, for example, in event-processing components because
it supports processing events with "exactly-once" semantics.

The example below uses 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"
)

# Generate new events. Typically, domain events of different
# types are generated in a domain model, and then serialized
# into NewEvent objects. An aggregate ID may be used as the
# name of a stream in KurrentDB.

stream_name1 = str(uuid.uuid4())
event1 = NewEvent(
    type='OrderCreated',
    data=b'{"order_number": "123456"}'
)
event2 = NewEvent(
    type='OrderSubmitted',
    data=b'{}'
)
event3 = NewEvent(
    type='OrderCancelled',
    data=b'{}'
)

# Append new events to a new stream. The value returned
# from the append_to_stream() method is the overall
# "commit position" in the database of the last new event
# recorded by this operation. The returned "commit position"
# may be used in a user interface to poll an eventually
# consistent event-processing component until it can
# present an up-to-date materialized view. New events are
# each allocated a "stream position", which is the next
# available position in the stream, starting from 0.

commit_position1 = client.append_to_stream(
    stream_name=stream_name1,
    current_version=StreamState.NO_STREAM,
    events=[event1, event2],
)

# Append events to an existing stream. The "current version"
# is the "stream position" of the last recorded event in a
# stream. We have recorded two new events, so the "current
# version" is 1. The exception 'WrongCurrentVersionError' will be
# raised if an incorrect value is given.

commit_position2 = client.append_to_stream(
    stream_name=stream_name1,
    current_version=1,
    events=[event3],
)

# - allocated commit positions increase monotonically
assert commit_position2 > commit_position1

# Get events recorded in a stream. This method returns
# a sequence of recorded event objects. The recorded
# event objects may be deserialized to domain event
# objects of different types and used to reconstruct
# an aggregate in a domain model.

recorded_events = client.get_stream(
    stream_name=stream_name1
)

# - stream 'stream_name1' now has three events
assert len(recorded_events) == 3

# - allocated stream positions are zero-based and gapless
assert recorded_events[0].stream_position == 0
assert recorded_events[1].stream_position == 1
assert recorded_events[2].stream_position == 2

# - event attribute values are recorded faithfully
assert recorded_events[0].type == "OrderCreated"
assert recorded_events[0].data == b'{"order_number": "123456"}'
assert recorded_events[0].id == event1.id

assert recorded_events[1].type == "OrderSubmitted"
assert recorded_events[1].data == b'{}'
assert recorded_events[1].id == event2.id

assert recorded_events[2].type == "OrderCancelled"
assert recorded_events[2].data == b'{}'
assert recorded_events[2].id == event3.id

# 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
# can be cancelled cleanly in case of any error.

received_events = []
with client.subscribe_to_all(commit_position=0) as subscription:
    # Iterate over the catch-up subscription. Process each recorded
    # event in turn. Within an atomic database transaction, record
    # the event's "commit position" along with any new state generated
    # by processing the event. Use the component's last recorded commit
    # position when restarting the catch-up subscription.

    for event in subscription:
        received_events.append(event)

        if event.commit_position == commit_position2:
            # Break so we can continue with the example.
            break

# - events are received in the order they were recorded
assert received_events[-3].type == "OrderCreated"
assert received_events[-3].data == b'{"order_number": "123456"}'
assert received_events[-3].id == event1.id

assert received_events[-2].type == "OrderSubmitted"
assert received_events[-2].data == b'{}'
assert received_events[-2].id == event2.id

assert received_events[-1].type == "OrderCancelled"
assert received_events[-1].data == b'{}'
assert received_events[-1].id == event3.id

# Close the client's gRPC connection.

client.close()
```


## Install package<a id="install-package"></a>

It is recommended to install Python packages into a Python virtual environment.

### From PyPI<a id="from-pypi"></a>

You can use pip to install this package directly from
[the Python Package Index](https://pypi.org/project/kurrentdbclient/).

    $ pip install kurrentdbclient

### With Poetry<a id="with-poetry"></a>

You can use Poetry to add this package to your pyproject.toml and install it.

    $ poetry add kurrentdbclient

## KurrentDB server<a id="kurrentdb-server"></a>

The KurrentDB server can be run locally using the official Docker container image.

### Run container<a id="run-container"></a>

For development, you can run a "secure" KurrentDB server using the following command.

    $ docker run -d --name kurrentdb-secure -it -p 2113:2113 --env "HOME=/tmp" docker.eventstore.com/kurrent-latest/kurrentdb:25.0.0-x64-8.0-bookworm-slim --dev

As we will see, your client will need a KurrentDB connection string URI as the value
of its `uri` constructor argument. The connection string for this "secure" KurrentDB
server would be:

    kurrentdb://admin:changeit@localhost:2113

To connect to a "secure" server, you will usually need to include a "username"
and a "password" in the connection string, so that the server can authenticate the
client. With KurrentDB, the default username is "admin" and the default password
is "changeit".

When connecting to a "secure" server, you may also need to provide an SSL/TLS certificate
as the value of the `root_certificates` constructor argument. If the server certificate
is publicly signed, the root certificates of the certificate authority may be installed
locally and picked up by the grpc package from a default location. The client uses the
root SSL/TLS certificate to authenticate the server. For development, you can either
use the SSL/TLS certificate of a self-signing certificate authority used to create the
server's certificate. Or, when using a single-node cluster, you can just use the server
certificate itself, getting the server certificate with the following Python code.

```python
import ssl

server_certificate = ssl.get_server_certificate(addr=('localhost', 2113))
```

Alternatively, you can start an "insecure" server using the following command.

    $ docker run -d --name kurrentdb-insecure -it -p 2113:2113 docker.eventstore.com/kurrent-latest/kurrentdb:25.0.0-x64-8.0-bookworm-slim --insecure

The connection string URI for this "insecure" server would be:

    kurrentdb://localhost:2113?Tls=false

As we will see, when connecting to an "insecure" server, there is no need to include
a "username" and a "password" in the connection string. If you do, these values will
be ignored by the client, so that they are not sent over an insecure channel.

Please note, the "insecure" connection string uses a query string with the field-value
`Tls=false`. The value of this field is by default `true`.

### Stop container<a id="stop-container"></a>

To stop and remove the "secure" container, use the following Docker commands.

    $ docker stop kurrentdb-secure
	$ docker rm kurrentdb-secure

To stop and remove the "insecure" container, use the following Docker commands.

    $ docker stop kurrentdb-insecure
	$ docker rm kurrentdb-insecure


## KurrentDB client<a id="kurrentdb-client"></a>

This KurrentDB client is implemented in the `kurrentdbclient` package with
the `KurrentDBClient` class.

### Import class<a id="import-class"></a>

The `KurrentDBClient` class can be imported from the `kurrentdbclient` package.

```python
from kurrentdbclient import KurrentDBClient
```

### Construct client<a id="construct-client"></a>

The `KurrentDBClient` class has one required constructor argument, `uri`, and three
optional constructor argument, `root_certificates`, `private_key`, and `certificate_chain`.

The `uri` argument is expected to be a KurrentDB connection string URI that
conforms with the standard KurrentDB "kurrentdb" or "kurrentdb+discover" URI schemes.

The client must be configured to create a "secure" connection to a "secure" server,
or alternatively an "insecure" connection to an "insecure" server. By default, the
client will attempt to create a "secure" connection. And so, when connecting to an
"insecure" server, the connection string must specify that the client should attempt
to make an "insecure" connection by using the URI query string field-value `Tls=false`.

The optional `root_certificates` argument can be either a Python `str` or a Python `bytes`
object containing PEM encoded SSL/TLS certificate(s), and is used to authenticate the
server to the client. When connecting to an "insecure" service, the value of this
argument will be ignored. When connecting to a "secure" server, it may be necessary to
set this argument. Typically, the value of this argument would be the public certificate
of the certificate authority that was responsible for generating the certificate used by
the KurrentDB server. It is unnecessary to set this value in this case if certificate
authority certificates are installed locally, such that the Python grpc library can pick
them up from a default location. Alternatively, for development, you can use the server's
certificate itself. The value of this argument is passed directly to `grpc.ssl_channel_credentials()`.

An alternative way to supply the `root_certificates` argument is through the `tlsCaFile` field-value of the connection string URI query string (see below). If the `tlsCaFile` field-value is specified, the `root_certificates` argument will be ignored.

The optional `private_key` and `certificate_chain` arguments are both either a Python
`str` or a Python `bytes` object. These arguments may be used to authenticate the client
to the server. It is necessary to provide correct values for these arguments when connecting
to a "secure" server that is running the commercial edition of KurrentDB with the
User Certificates plugin enabled. The value of `private_key` should be the X.509 user
certificate's private key in PEM format. The value of `certificate_chain` should be the
X.509 user certificate itself in PEM format. The values of these arguments are passed
directly to `grpc.ssl_channel_credentials()`. When connecting to an "insecure" service,
the values of these arguments will be ignored. Please note, an alternative way of
supplying the client with a user certificate and private key is to use the `UserCertFile`
and `UserKeyFile` field-values of the connection string URI query string (see below).
If the `UserCertFile` field-value is specified, the `certificate_chain` argument will be
ignored. If the `UserKeyFile` field-value is specified, the `public_key` argument will be
ignored.

In the example below, constructor argument values for `uri` and `root_certificates` are
taken from the operating system environment.

```python
import os

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

## Connection strings<a id="connection-strings"></a>

A KurrentDB connection string is a URI that conforms with one of two possible
schemes: either the "kurrentdb" scheme, or the "kurrentdb+discover" scheme.

The syntax and semantics of the KurrentDB URI schemes are described below. The
syntax is defined using [EBNF](https://en.wikipedia.org/wiki/Extended_Backus–Naur_form).

Please note, also supported are "kdb" and "esdb" which are synonyms for the "kurrentdb"
scheme, and "kdb+discover" and "esdb+disover" which are synonyms for the "kdb+discover"
scheme.

### Two schemes<a id="two-schemes"></a>

The "kurrentdb" URI scheme can be defined in the following way.

    kurrentdb-uri = "kurrentdb://" , [ user-info , "@" ] , grpc-target, { "," , grpc-target } , [ "?" , query-string ] ;

In the "kurrentdb" URI scheme, after the optional user info string, there must be at least
one gRPC target. If there are several gRPC targets, they must be separated from each
other with the "," character.

Each gRPC target should indicate a KurrentDB gRPC server socket, all in the same
KurrentDB cluster, by specifying a host and a port number separated with the ":"
character. The host may be a hostname that can be resolved to an IP address, or an IP
address.

    grpc-target = ( hostname | ip-address ) , ":" , port-number ;

If there is one gRPC target, the client will simply attempt to connect to this
server, and it will use this connection when recording and retrieving events.

If there are two or more gRPC targets, the client will attempt to connect to the
Gossip API of each in turn, attempting to obtain information about the cluster from
it, until information about the cluster is obtained. A member of the cluster is then
selected by the client according to the "node preference" specified by the connection
string URI. The client will then close its connection and connect to the selected node
without the 'round robin' load balancing strategy. If the "node preference" is "leader",
and after connecting to a leader, if the leader becomes a follower, the client will
reconnect to the new leader.


The "kurrentdb+discover" URI scheme can be defined in the following way.

    kurrentdb-discover-uri = "kurrentdb+discover://" , [ user-info, "@" ] , cluster-domainname, [ ":" , port-number ] , [ "?" , query-string ] ;

In the "kurrentdb+discover" URI scheme, after the optional user info string, there should be
a domain name which identifies a cluster of KurrentDB servers. Individual nodes in
the cluster should be declared with DNS 'A' records.

The client will use the cluster domain name with the gRPC library's 'round robin' load
balancing strategy to call the Gossip APIs of addresses discovered from DNS 'A' records.
Information about the KurrentDB cluster is obtained from the Gossip API. A member of
the cluster is then selected by the client according to the "node preference" option.
The client will then close its connection and connect to the selected node without the
'round robin' load balancing strategy. If the "node preference" is "leader",
and after connecting to a leader, if the leader becomes a follower, the client will
reconnect to the new leader.

### User info string<a id="user-info-string"></a>

In both the "kurrentdb" and "kurrentdb+discover" schemes, the URI may include a user info string.
If it exists in the URI, the user info string must be separated from the rest of the URI
with the "@" character. The user info string must include a username and a password,
separated with the ":" character.

    user-info = username , ":" , password ;

The user info is sent by the client in a "basic auth" authorization header in each gRPC
call to a "secure" server. This authorization header is used by the server to authenticate
the client. The Python gRPC library does not allow call credentials to be transferred to
"insecure" servers.

### Query string<a id="query-string"></a>

In both the "kurrentdb" and "kurrentdb+discover" schemes, the optional query string must be one
or many field-value arguments, separated from each other with the "&" character.

    query-string = field-value, { "&", field-value } ;

Each field-value argument must be one of the supported fields, and an
appropriate value, separated with the "=" character.

    field-value = ( "Tls", "=" , "true" | "false" )
                | ( "TlsVerifyCert", "=" , "true" | "false" )
                | ( "ConnectionName", "=" , string )
                | ( "NodePreference", "=" , "leader" | "follower" | "readonlyreplica" | "random" )
                | ( "DefaultDeadline", "=" , integer )
                | ( "GossipTimeout", "=" , integer )
                | ( "MaxDiscoverAttempts", "=" , integer )
                | ( "DiscoveryInterval", "=" , integer )
                | ( "KeepAliveInterval", "=" , integer )
                | ( "KeepAliveTimeout", "=" , integer ) ;
                | ( "TlsCaFile", "=" , string ) ;
                | ( "UserCertFile", "=" , string ) ;
                | ( "UserKeyFile", "=" , string ) ;

The table below describes the query string field-values supported by this client.

| Field               | Value                                                                 | Description                                                                                                                                                       |
|---------------------|-----------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Tls                 | "true", "false" (default: "true")                                     | If "true" the client will create a "secure" gRPC channel. If "false" the client will create an "insecure" gRPC channel. This must match the server configuration. |
| TlsVerifyCert       | "true", "false" (default: "true")                                     | This value is currently ignored.                                                                                                                                  |
| ConnectionName      | string (default: auto-generated version-4 UUID)                       | Sent in call metadata for every call, to identify the client to the cluster.                                                                                      |
| NodePreference      | "leader", "follower", "readonlyreplica", "random" (default: "leader") | The node state preferred by the client. The client will select a node from the cluster info received from the Gossip API according to this preference.            |
| DefaultDeadline     | integer (default: `None`)                                             | The default value (in seconds) of the `timeout` argument of client "write" methods such as `append_to_stream()`.                                                  |
| GossipTimeout       | integer (default: 5)                                                  | The default value (in seconds) of the `timeout` argument of gossip read methods, such as `read_gossip()`.                                                         |
| MaxDiscoverAttempts | integer (default: 10)                                                 | The number of attempts to read gossip when connecting or reconnecting to a cluster member.                                                                        |
| DiscoveryInterval   | integer (default: 100)                                                | How long to wait (in milliseconds) between gossip retries.                                                                                                        |
| KeepAliveInterval   | integer (default: `None`)                                             | The value (in milliseconds) of the "grpc.keepalive_ms" gRPC channel option.                                                                                       |
| KeepAliveTimeout    | integer (default: `None`)                                             | The value (in milliseconds) of the "grpc.keepalive_timeout_ms" gRPC channel option.                                                                               |
| TlsCaFile           | string (default: `None`)                                              | Absolute filesystem path to file containing the CA certicate in PEM format. This will be used to verify the server's certificate.                                 |
| UserCertFile        | string (default: `None`)                                              | Absolute filesystem path to file containing the X.509 user certificate in PEM format.                                                                             |
| UserKeyFile         | string (default: `None`)                                              | Absolute filesystem path to file containing the X.509 user certificate's private key in PEM format.                                                               |


Please note, the client is insensitive to the case of fields and values. If fields are
repeated in the query string, the query string will be parsed without error. However,
the connection options used by the client will use the value of the first field. All
the other field-values in the query string with the same field name will be ignored.
Fields without values will also be ignored.

If the client's node preference is "follower" and there are no follower
nodes in the cluster, then the client will raise an exception. Similarly, if the
client's node preference is "readonlyreplica" and there are no read-only replica
nodes in the cluster, then the client will also raise an exception.

The gRPC channel option "grpc.max_receive_message_length" is automatically
configured to the value `17 * 1024 * 1024`. This value cannot be configured.


### Examples<a id="examples"></a>

Here are some examples of KurrentDB connection string URIs.

The following URI will cause the client to make an "insecure" connection to
gRPC target `'localhost:2113'`. Because the client's node preference is "follower",
methods that can be called on a follower should complete successfully, methods that
require a leader will raise a `NodeIsNotLeaderError` exception.

    kurrentdb://127.0.0.1:2113?Tls=false&NodePreference=follower

The following URI will cause the client to make an "insecure" connection to
gRPC target `'localhost:2113'`. Because the client's node preference is "leader",
if this node is not a leader, then a `NodeIsNotLeaderError` exception will be raised by
all methods.

    kurrentdb://127.0.0.1:2113?Tls=false&NodePreference=leader

The following URI will cause the client to make a "secure" connection to
gRPC target `'localhost:2113'` with username `'admin'` and password `'changeit'`
as the default call credentials when making calls to the KurrentDB gRPC API.
Because the client's node preference is "leader", by default, if this node is not
a leader, then a `NodeIsNotLeaderError` exception will be raised by all methods.

    kurrentdb://admin:changeit@localhost:2113

The following URI will cause the client to make "secure" connections, firstly to
get cluster info from either `'localhost:2111'`, or `'localhost:2112'`, or `'localhost:2113'`.
Because the client's node preference is "leader", the client will select the leader
node from the cluster info and reconnect to the leader. If the "leader" node becomes
a "follower" and another node becomes "leader", then the client will reconnect to the
new leader.

    kurrentdb://admin:changeit@localhost:2111,localhost:2112,localhost:2113?NodePreference=leader


The following URI will cause the client to make "secure" connections, firstly to
get cluster info from either `'localhost:2111'`, or `'localhost:2112'`, or `'localhost:2113'`.
Because the client's node preference is "follower", the client will select a follower
node from the cluster info and reconnect to this follower. Please note, if the "follower"
node becomes the "leader", the client will not reconnect to a follower -- such behavior
may be implemented in a future version of the client and server.

    kurrentdb://admin:changeit@localhost:2111,localhost:2112,localhost:2113?NodePreference=follower


The following URI will cause the client to make "secure" connections, firstly to get
cluster info from addresses in DNS 'A' records for `'cluster1.example.com'`, and then
to connect to a "leader" node. The client will use a default timeout
of 5 seconds when making calls to KurrentDB API "write" methods.

    kurrentdb+discover://admin:changeit@cluster1.example.com?DefaultDeadline=5


The following URI will cause the client to make "secure" connections, firstly to get
cluster info from addresses in DNS 'A' records for `'cluster1.example.com'`, and then
to connect to a "leader" node. It will configure gRPC connections with a "keep alive
interval" and a "keep alive timeout".

    kurrentdb+discover://admin:changeit@cluster1.example.com?KeepAliveInterval=10000&KeepAliveTimeout=10000


## Event objects<a id="event-objects"></a>

This package defines a `NewEvent` class and a `RecordedEvent` class. The
`NewEvent` class should be used when writing events to the database. The
`RecordedEvent` class is used when reading events from the database.

### New events<a id="new-events"></a>

The `NewEvent` class should be used when writing events to a KurrentDB database.
You will need to construct new event objects before calling `append_to_stream()`.

The `NewEvent` class is a frozen Python dataclass. It has two required constructor
arguments (`type` and `data`) and three optional constructor arguments (`metadata`,
`content_type` and `id`).

The required `type` argument is a Python `str`, used to describe the type of
domain event that is being recorded.

The required `data` argument is a Python `bytes` object, used to state the
serialized data of the domain event that is being recorded.

The optional `metadata` argument is a Python `bytes` object, used to indicate any
metadata of the event that will be recorded. The default value is an empty `bytes`
object.

The optional `content_type` argument is a Python `str`, used to indicate the
kind of data that is being recorded. The default value is `'application/json'`,
which indicates that the `data` was serialised using JSON. An alternative value
for this argument is the more general indication `'application/octet-stream'`.

The optional `id` argument is a Python `UUID` object, used to specify the unique ID
of the event that will be recorded. If no value is provided, a new version-4 UUID
will be generated.

```python
new_event1 = NewEvent(
    type='OrderCreated',
    data=b'{"name": "Greg"}',
)
assert new_event1.type == 'OrderCreated'
assert new_event1.data == b'{"name": "Greg"}'
assert new_event1.metadata == b''
assert new_event1.content_type == 'application/json'
assert isinstance(new_event1.id, uuid.UUID)

event_id = uuid.uuid4()
new_event2 = NewEvent(
    type='ImageCreated',
    data=b'01010101010101',
    metadata=b'{"a": 1}',
    content_type='application/octet-stream',
    id=event_id,
)
assert new_event2.type == 'ImageCreated'
assert new_event2.data == b'01010101010101'
assert new_event2.metadata == b'{"a": 1}'
assert new_event2.content_type == 'application/octet-stream'
assert new_event2.id == event_id
```

### Recorded events<a id="recorded-events"></a>

The `RecordedEvent` class is used when reading events from a KurrentDB
database. The client will return event objects of this type from all methods
that return recorded events, such as `get_stream()`, `subscribe_to_all()`,
and `read_subscription_to_all()`. You do not need to construct recorded event objects.

Like `NewEvent`, the `RecordedEvent` class is a frozen Python dataclass. It has
all the attributes that `NewEvent` has (`type`, `data`, `metadata`, `content_type`, `id`)
that follow from an event having been recorded, and some additional attributes that follow
from the recording of an event (`stream_name`, `stream_position`, `commit_position`,
`recorded_at`). It also has a `link` attribute, which is `None` unless the recorded
event is a "link event" that has been "resolved" to the linked event. And it has a
`retry_count` which has an integer value when receiving recorded events from persistence
subscriptions, otherwise the value of `retry_count` is `None`.

The `type` attribute is a Python `str`, used to indicate the type of an event
that was recorded.

The `data` attribute is a Python `bytes` object, used to indicate the data of an
event that was recorded.

The `metadata` attribute is a Python `bytes` object, used to indicate the metadata of
an event that was recorded.

The `content_type` attribute is a Python `str`, used to indicate the type of
data that was recorded for an event. It is usually `'application/json'`, indicating
that the data can be parsed as JSON. Alternatively, it is `'application/octet-stream'`.

The `id` attribute is a Python `UUID` object, used to indicate the unique ID of an
event that was recorded.

The `stream_name` attribute is a Python `str`, used to indicate the name of a
stream in which an event was recorded.

The `stream_position` attribute is a Python `int`, used to indicate the position in a
stream at which an event was recorded.

In KurrentDB, a "stream position" is an integer representing the position of a
recorded event in a stream. Each recorded event is recorded at a position in a stream.
Each stream position is occupied by only one recorded event. New events are recorded at the
next unoccupied position. All sequences of stream positions are zero-based and gapless.

The `commit_position` attribute is a Python `int`, used to indicate the position in the
database at which an event was recorded.

In KurrentDB, a "commit position" is an integer representing the position of a
recorded event in the database. Each recorded event is recorded at a position in the
database. Each commit position is occupied by only one recorded event. Commit positions
are zero-based and increase monotonically as new events are recorded. But, unlike stream
positions, the sequence of successive commit positions is not gapless. Indeed, there are
usually large differences between the commit positions of successively recorded events.

The `recorded_at` attribute is a Python `datetime`, used to indicate when an event was
recorded by the database.

The `link` attribute is an optional `RecordedEvent` that carries information about
a "link event" that has been "resolved" to the linked event. This allows access to
the link event attributes when link events have been resolved, for example access
to the correct event ID to be used when acknowledging or negatively acknowledging
link events. Link events are "resolved" when the `resolve_links` argument is `True`
and when replaying parked events (negatively acknowledging an event received from
a persistent subscription with the `'park'` action will create a link event, and
when parked events are replayed they are received as resolved events). The
`ack_id` property helps with obtaining the correct event ID to use when acknowledging
or negatively acknowledging events received from persistent subscriptions.

The `retry_count` is a Python `int`, used to indicate the number of times a persistent
subscription has retried sending the event to a consumer.

```python
from dataclasses import dataclass
from datetime import datetime

@dataclass(frozen=True)
class RecordedEvent:
    """
    Encapsulates event data that has been recorded in KurrentDB.
    """

    type: str
    data: bytes
    metadata: bytes
    content_type: str
    id: UUID
    stream_name: str
    stream_position: int
    commit_position: Optional[int]
    recorded_at: Optional[datetime] = None
    link: Optional["RecordedEvent"] = None
    retry_count: Optional[int] = None

    @property
    def ack_id(self) -> UUID:
        if self.link is not None:
            return self.link.id
        else:
            return self.id

    @property
    def is_system_event(self) -> bool:
        return self.type.startswith("$")

    @property
    def is_link_event(self) -> bool:
        return self.type == "$>"

    @property
    def is_resolved_event(self) -> bool:
        return self.link is not None

    @property
    def is_checkpoint(self) -> bool:
        return False

    @property
    def is_caught_up(self) -> bool:
        return False

    @property
    def is_fell_behind(self) -> bool:
        return False
```

The property `ack_id` can be used to obtain the correct event ID to `ack()` or `nack()`
events received when reading persistent subscriptions. The returned value is either the
value of the `id` attribute of the `link` attribute, if `link` is not `None`, otherwise
it is the value of the `id` attribute.

The property `is_system_event` indicates whether the event is a "system event". System
events have a `type` value that starts with `'$'`.

The property `is_link_event` indicates whether the event is a "link event". Link
events have a `type` value of `'$>'`.

The property `is_resolve_event` indicates whether the event has been resolved from a
"link event". The returned value is `True` if `link` is not `None`.

The property `is_checkpoint` is `False`. This can be used to identify `Checkpoint`
instances returned when subscribing to events with `includ