Python SDK

Official SDK

This SDK is officially developed and maintained by the native web GmbH, the creators of EventSourcingDB. It is aligned with the latest platform features and receives regular updates.

Installation#

To install the SDK, run:

pip install eventsourcingdb

You can also find this SDK on PyPI.

Capabilities#

This SDK intentionally focuses on application-level use cases. It provides core functionality for writing, reading, and observing events, and deliberately excludes operational concerns such as backup, restore, or health monitoring.

For a detailed explanation of the specific requirements behind each capability, see the Compliance Criteria.

Compatibility

Python SDK 1.11 runs with EventSourcingDB 1.1 and later.

Essentials

✓ Initialize the Client
✓ Ping the Server
✓ Verify API Token
Added in 1.11✓ Detect a Stalled Connection
✓ Abort a Request

Writing Events

✓ Write Single Events
✓ Write Multiple Events
✓ Use isSubjectPristine
Added in 1.8Requires EventSourcingDB 1.2✓ Use isSubjectPopulated
✓ Use isSubjectOnEventId
Added in 1.4✓ Use isEventQlQueryTrue
✓ Include OpenTelemetry Trace Context

Reading Events

✓ Read Events by Subject
✓ Read Recursively
✓ Read Chronological and Anti-Chronological
✓ Use Boundaries
✓ Use fromLatestEvent

Using EventQL

✓ Execute EventQL Queries

Observing Events

✓ Observe Events By Subject
✓ Observe Recursively
✓ Use Boundaries
✓ Use fromLatestEvent

Metadata and Discovery

✓ Register Event Schemas
✓ List Subjects
✓ List Event Types
Added in 1.5✓ List a Specific Event Type
Added in 1.6✓ Verify Event Hashes
Added in 1.7✓ Verify Event Signatures

Testcontainers Support

✓ Start and Stop EventSourcingDB in Tests
✓ Use Custom Configuration (e.g., API Token, Port, …)

Getting Started#

Import the Client class and create an instance by providing the URL of your EventSourcingDB instance and the API token to use:

from eventsourcingdb import Client

url = 'http://localhost:3000'
api_token = 'secret'

client = Client(
  base_url = url,
  api_token = api_token,
)

Then call the ping function to check whether the instance is reachable. If it is not, the function will raise an error:

await client.ping()

Note that ping does not require authentication, so the call may succeed even if the API token is invalid.

If you want to verify the API token, call verify_api_token. If the token is invalid, the function will raise an error:

await client.verify_api_token()

Configuring the Stall TimeoutAdded in 1.11#

A request fails once nothing at all has arrived from the server for 30 seconds, measured from sending the request until its end, including connecting and waiting for the response. Everything that arrives, such as a part of a response, an event, a row, or a heartbeat, restarts the clock, so a long-running request that keeps receiving data is never cut off. The time your code spends between two items of a stream does not count. To change the limit, pass the stall_timeout keyword argument in seconds when creating the client:

client = Client(
  base_url = url,
  api_token = api_token,
  stall_timeout = 10,
)

The stall timeout cannot be turned off: it must be a finite number of seconds greater than zero, otherwise creating the client raises a ValueError.

Added in 1.11Changed in 2.0What a request raises once the server stalls depends on the request:

If connecting to the server stalls, the request raises aiohttp's ConnectionTimeoutError. read_event_types wraps it in an InternalError, as it does with every error of the connection.
ping, verify_api_token, write_events, register_event_schema, and read_event_type raise a TimeoutError.
read_events, read_subjects, and read_event_types raise a ConnectionStalledError, which is a subclass of ServerError.
observe_events and run_eventql_query raise a HeartbeatTimeoutError once the server has answered, and a ConnectionStalledError before. HeartbeatTimeoutError is a subclass of ConnectionStalledError. For details, see Detecting a Stalled Connection.

The stall timeout also covers the time a request waits for a free connection: a client keeps at most 100 connections open at the same time, so with as many streams open, a further request waits until one of them ends. If it waits for longer than the stall timeout, it fails as if the server had stalled after the request was sent, with a TimeoutError or a ConnectionStalledError as listed above.

In any case, the client closes the connection. Your own timeouts and cancellations, such as asyncio.timeout, still apply and can end a request earlier.

Writing Events#

Call the write_events function and hand over a list with one or more event candidates. You do not have to provide all event fields – some are automatically added by the server.

Specify source, subject, type, and data according to the CloudEvents format. Optionally, specify trace_parent and trace_state.

The function returns the written events, including the fields added by the server:

from eventsourcingdb import EventCandidate

written_events = await client.write_events(
  event_candidates = [
    EventCandidate(
      source = 'https://library.eventsourcingdb.io',
      subject = '/books/42',
      type = 'io.eventsourcingdb.library.book-acquired',
      data = {
        'title': '2001 – A Space Odyssey',
        'author': 'Arthur C. Clarke',
        'isbn': '978-0756906788',
      },
    )
  ]
)

Using the `isSubjectPristine` precondition#

If you only want to write events in case a subject (such as /books/42) does not yet have any events, import the IsSubjectPristine class and pass it as the second argument as a list of preconditions:

from eventsourcingdb import IsSubjectPristine

written_events = await client.write_events(
  event_candidates = [
    # event candidates
  ],
  preconditions = [
    IsSubjectPristine('/books/42')
  ],
)

Using the `isSubjectPopulated` preconditionAdded in 1.8Requires EventSourcingDB 1.2#

If you only want to write events in case a subject (such as /books/42) already has at least one event, import the IsSubjectPopulated class and pass it as the second argument as a list of preconditions:

from eventsourcingdb import IsSubjectPopulated

written_events = await client.write_events(
  event_candidates = [
    # event candidates
  ],
  preconditions = [
    IsSubjectPopulated('/books/42')
  ],
)

Using the `isSubjectOnEventId` precondition#

If you only want to write events in case the last event of a subject (such as /books/42) has a specific ID (e.g., 0), import the IsSubjectOnEventId class and pass it as a list of preconditions in the second argument:

from eventsourcingdb import IsSubjectOnEventId

written_events = await client.write_events(
  event_candidates = [
    # event candidates
  ],
  preconditions = [
    IsSubjectOnEventId('/books/42', '0')
  ],
)

Note that according to the CloudEvents standard, event IDs must be of type string.

Using the `isEventQlQueryTrue` preconditionChanged in 1.5#

If you want to write events depending on an EventQL query, import the IsEventQlQueryTrue class and pass it as a list of preconditions in the second argument:

from eventsourcingdb import IsEventQlQueryTrue

written_events = await client.write_events(
  event_candidates = [
    # event candidates
  ],
  preconditions = [
    IsEventQlQueryTrue('FROM e IN events WHERE e.type == "io.eventsourcingdb.library.book-borrowed" PROJECT INTO COUNT() < 10')
  ],
)

Note that the query must return a single row with a single value, which is interpreted as a boolean.

Reading Events#

To read all events of a subject, call the read_events function with the subject as the first argument and an options object as the second argument. Set the recursive option to False. This ensures that only events of the given subject are returned, not events of nested subjects.

The function returns an asynchronous generator, which you can use e.g. inside an async for loop:

from eventsourcingdb import ReadEventsOptions

async for event in client.read_events(
  subject = '/books/42',
  options = ReadEventsOptions(
    recursive = False
  ),
):
  pass

Reading From Subjects Recursively#

If you want to read not only all the events of a subject, but also the events of all nested subjects, set the recursive option to True:

from eventsourcingdb import ReadEventsOptions

async for event in client.read_events(
  subject = '/books/42',
  options = ReadEventsOptions(
    recursive = True
  ),
):
  pass

This also allows you to read all events ever written. To do so, provide / as the subject and set recursive to True, since all subjects are nested under the root subject.

Reading in Anti-Chronological Order#

By default, events are read in chronological order. To read in anti-chronological order, provide the order option and set it to Order.ANTICHRONOLOGICAL:

from eventsourcingdb import (
  Order,
  ReadEventsOptions,
)

async for event in client.read_events(
  subject = '/books/42',
  options = ReadEventsOptions(
    recursive = False,
    order = Order.ANTICHRONOLOGICAL,
  ),
):
  pass

Note that you can also specify Order.CHRONOLOGICAL to explicitly enforce the default order.

Specifying Bounds#

Sometimes you do not want to read all events, but only a range of events. For that, you can specify the lower_bound and upper_bound options – either one of them or even both at the same time.

Specify the ID and whether to include or exclude it, for both the lower and upper bound:

from eventsourcingdb import (
  Bound,
  BoundType,
  ReadEventsOptions,
)

async for event in client.read_events(
  subject = '/books/42',
  options = ReadEventsOptions(
    recursive = False,
    lower_bound = Bound(id = '100', type = BoundType.INCLUSIVE),
    upper_bound = Bound(id = '200', type = BoundType.EXCLUSIVE),
  ),
):
  pass

Starting From the Latest Event of a Given Type#

To read starting from the latest event of a given type, provide the from_latest_event option and specify the subject, the type, and how to proceed if no such event exists.

Possible options are IfEventIsMissingDuringRead.READ_NOTHING, which skips reading entirely, or IfEventIsMissingDuringRead.READ_EVERYTHING, which effectively behaves as if from_latest_event was not specified:

from eventsourcingdb import (
  IfEventIsMissingDuringRead,
  ReadEventsOptions,
  ReadFromLatestEvent,
)

async for event in client.read_events(
  subject = '/books/42',
  options = ReadEventsOptions(
    recursive = False,
    from_latest_event = ReadFromLatestEvent(
      subject = '/books/42',
      type = 'io.eventsourcingdb.library.book-borrowed',
      if_event_is_missing = IfEventIsMissingDuringRead.READ_EVERYTHING,
    ),
  ),
):
  pass

Note that from_latest_event and lower_bound can not be provided at the same time.

Aborting Reading#

If you need to abort reading use break or return within the async for loop. However, this only works if there is currently an iteration going on.

To abort reading independently of that, store the generator in a variable, and close it explicitly:

from eventsourcingdb import ReadEventsOptions

events = client.read_events(
  subject = '/books/42',
  options = ReadEventsOptions(
    recursive = False,
  ),
)

async for event in events:
  pass

# Somewhere else, abort the generator, which will cause
# reading to end.
await events.aclose()

Running EventQL Queries#

To run an EventQL query, call the run_eventql_query function and provide the query as argument. The function returns an asynchronous generator, which you can use e.g. inside an async for loop:

async for row in client.run_eventql_query(
  query = '''
    FROM e IN events
    PROJECT INTO e
  ''',
):
  pass

Note that each row returned by the generator matches the projection specified in your query.

Changed in 1.11Changed in 2.0While the query runs, the server sends a heartbeat every second whenever there is no row to send. If neither a row nor a heartbeat arrives for the stall timeout, 30 seconds by default, the generator closes the connection and raises a HeartbeatTimeoutError, which is a subclass of ConnectionStalledError and of ServerError. For details, see Detecting a Stalled Connection.

Aborting a Query#

If you need to abort a query use break or return within the async for loop. However, this only works if there is currently an iteration going on.

To abort the query independently of that, store the generator in a variable, and close it explicitly:

rows = client.run_eventql_query(
  query = '''
    FROM e IN events
    PROJECT INTO e
  ''',
)

async for row in rows:
  pass

# Somewhere else, abort the generator, which will cause
# the query to end.
await rows.aclose()

Converting Events to pandas DataFrameAdded in 1.9#

Added in 1.9Changed in 2.3For data analysis and exploration, you can convert event streams to pandas DataFrames. The client SDK does not install pandas itself, so to use this feature, install pandas next to it:

pip install eventsourcingdb pandas

Added in 1.9Import the events_to_dataframe function from the eventsourcingdb.pandas module and pass an event stream to it:

from eventsourcingdb import Client, ReadEventsOptions
from eventsourcingdb.pandas import events_to_dataframe

events = client.read_events(
  subject = '/books',
  options = ReadEventsOptions(
    recursive = True
  ),
)

df = await events_to_dataframe(events)

The resulting DataFrame includes all event fields as columns: event_id, time, source, subject, type, data, spec_version, data_content_type, predecessor_hash, hash, trace_parent, trace_state, and signature.

The data field remains as a dict column, which you can access directly:

# Access the data field
for index, row in df.iterrows():
  print(row['data'])

Flattening the Data FieldAdded in 1.9#

For analysis of specific event types, you may want to flatten the data field into separate columns. Use pandas' json_normalize function:

import pandas as pd

# Filter for a specific event type first
book_acquired_df = df[df['type'] == 'io.eventsourcingdb.library.book-acquired']

# Flatten the data field
flattened_df = book_acquired_df.join(
  pd.json_normalize(book_acquired_df['data'])
)

# Now you can access data fields as columns
print(flattened_df['title'])
print(flattened_df['author'])

Observing Events#

To observe all events of a subject, call the observe_events function with the subject as the first argument and an options object as the second argument. Set the recursive option to False. This ensures that only events of the given subject are returned, not events of nested subjects.

The function returns an asynchronous generator, which you can use e.g. inside an async for loop:

from eventsourcingdb import ObserveEventsOptions

async for event in client.observe_events(
  subject = '/books/42',
  options = ObserveEventsOptions(
    recursive = False
  ),
):
  pass

Observing From Subjects Recursively#

If you want to observe not only all the events of a subject, but also the events of all nested subjects, set the recursive option to True:

from eventsourcingdb import ObserveEventsOptions

async for event in client.observe_events(
  subject = '/books/42',
  options = ObserveEventsOptions(
    recursive = True
  ),
):
  pass

This also allows you to observe all events ever written. To do so, provide / as the subject and set recursive to True, since all subjects are nested under the root subject.

Specifying Bounds#

Sometimes you do not want to observe all events, but only a range of events. For that, you can specify the lower_bound option.

Specify the ID and whether to include or exclude it:

from eventsourcingdb import (
  Bound,
  BoundType,
  ObserveEventsOptions,
)

async for event in client.observe_events(
  subject = '/books/42',
  options = ObserveEventsOptions(
    recursive = False,
    lower_bound = Bound(id = '100', type = BoundType.INCLUSIVE),
  ),
):
  pass

Starting From the Latest Event of a Given Type#

To observe starting from the latest event of a given type, provide the from_latest_event option and specify the subject, the type, and how to proceed if no such event exists.

Possible options are IfEventIsMissingDuringObserve.WAIT_FOR_EVENT, which waits for an event of the given type to happen, or IfEventIsMissingDuringObserve.READ_EVERYTHING, which effectively behaves as if from_latest_event was not specified:

from eventsourcingdb import (
  IfEventIsMissingDuringObserve,
  ObserveEventsOptions,
  ObserveFromLatestEvent,
)

async for event in client.observe_events(
  subject = '/books/42',
  options = ObserveEventsOptions(
    recursive = False,
    from_latest_event = ObserveFromLatestEvent(
      subject = '/books/42',
      type = 'io.eventsourcingdb.library.book-borrowed',
      if_event_is_missing = IfEventIsMissingDuringObserve.READ_EVERYTHING,
    ),
  ),
):
  pass

Note that from_latest_event and lower_bound can not be provided at the same time.

Detecting a Stalled ConnectionChanged in 1.11Changed in 2.0#

While observing, the server sends a heartbeat every second whenever there is no new event. If neither an event nor a heartbeat arrives for the stall timeout, 30 seconds by default, for example because a proxy keeps the connection open but no longer passes anything on, the generator closes the connection and raises a HeartbeatTimeoutError. Since HeartbeatTimeoutError is a subclass of ConnectionStalledError, which in turn is a subclass of ServerError, you can handle it specifically, together with other stalled connections, or together with other server errors:

from eventsourcingdb import HeartbeatTimeoutError, ObserveEventsOptions

try:
  async for event in client.observe_events(
    subject = '/books/42',
    options = ObserveEventsOptions(
      recursive = False
    ),
  ):
    pass
except HeartbeatTimeoutError:
  # Neither an event nor a heartbeat arrived for the stall timeout.
  pass

Aborting Observing#

Added in 1.10Changed in 2.0Observing has no time limit of its own and runs until you abort it.

If you need to abort observing use break or return within the async for loop. However, this only works if there is currently an iteration going on.

To abort observing independently of that, store the generator in a variable, and close it explicitly:

from eventsourcingdb import ObserveEventsOptions

events = client.observe_events(
  subject = '/books/42',
  options = ObserveEventsOptions(
    recursive = False
  ),
)

async for event in events:
  pass

# Somewhere else, abort the generator, which will cause
# observing to end.
await events.aclose()

Registering an Event Schema#

To register an event schema, call the register_event_schema function and hand over an event type and the desired schema:

await client.register_event_schema(
  event_type = 'io.eventsourcingdb.library.book-acquired',
  json_schema = {
    'type': 'object',
    'properties': {
      'title': { 'type': 'string' },
      'author': { 'type': 'string' },
      'isbn': { 'type': 'string' },
    },
    'required': [
      'title',
      'author',
      'isbn'
    ],
    'additionalProperties': False,
  },
)

Listing Subjects#

Changed in 2.0To list all subjects, call the read_subjects function with / as the base subject. The function returns an asynchronous generator, which you can use e.g. inside an async for loop:

async for subject in client.read_subjects(
  base_subject = '/'
):
  pass

If you only want to list subjects within a specific branch, provide the desired base subject instead:

async for subject in client.read_subjects(
  base_subject = '/books'
):
  pass

Aborting Listing#

If you need to abort listing use break or return within the async for loop. However, this only works if there is currently an iteration going on.

To abort listing independently of that, store the generator in a variable, and close it explicitly:

Changed in 2.0

subjects = client.read_subjects(
  base_subject = '/'
)

async for subject in subjects:
  pass

# Somewhere else, abort the generator, which will cause
# reading to end.
await subjects.aclose()

Listing Event Types#

To list all event types, call the read_event_types function. The function returns an asynchronous generator, which you can use e.g. inside an async for loop:

async for event_type in client.read_event_types():
  pass

Aborting Listing#

If you need to abort listing use break or return within the async for loop. However, this only works if there is currently an iteration going on.

To abort listing independently of that, store the generator in a variable, and close it explicitly:

event_types = client.read_event_types()

async for event_type in event_types:
  pass

# Somewhere else, abort the generator, which will cause
# reading to end.
await event_types.aclose()

Listing a Specific Event TypeAdded in 1.5#

To list a specific event type, call the read_event_type function with the event type as an argument. The function returns the detailed event type, which includes the schema:

event_type = await client.read_event_type("io.eventsourcingdb.library.book-acquired")

Verifying an Event's HashChanged in 1.11#

To verify the integrity of an event, call the verify_hash function on the event instance. This recomputes the event's hash locally and compares it to the hash stored in the event. If the hashes differ, the function raises a HashVerificationFailedError:

event.verify_hash()

The hash of an event has a version. EventSourcingDB writes hashes of version 2, which start with esdb:hash:v2:, since version 1.3, while events written before keep their hashes of version 1, which have no prefix. The verify_hash function supports both versions and computes the hash over the event's data exactly as the server sent it. If the hash is of a version the client SDK does not support, the function raises an UnsupportedHashVersionError, whose version attribute names the version.

Added in 1.7Note that this only verifies the hash. If you also want to verify the signature, you can skip this step and call verify_signature directly, which performs a hash verification internally.

Verifying an Event's SignatureAdded in 1.7#

To verify the authenticity of an event, call the verify_signature function on the event instance. This requires the public key that matches the private key used for signing on the server.

The function first verifies the event's hash, and then checks the signature. If any verification step fails, it raises an error:

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

# ...

verification_key = # public key as Ed25519PublicKey

event.verify_signature(verification_key)

Added in 1.11If the hash verification fails, the function raises the same errors as verify_hash. If the signature does not match the verification key, or if it is malformed, the function raises a SignatureVerificationFailedError. If the event has no signature at all, for example because the server was started without a signing key, the function raises a ValidationError.

HashVerificationFailedError, UnsupportedHashVersionError, and SignatureVerificationFailedError are subclasses of ValidationError, so you can handle them specifically or together with other validation errors.

Using Testcontainers#

Import the Container class, create an instance, call the start function to run a test container, get a client, run your test code, and finally call the stop function to stop the test container:

from eventsourcingdb import Container

container = Container()
container.start()

client = container.get_client()

# ...

container.stop()

To check if the test container is running, call the is_running function:

is_running = container.is_running()

Configuring the Container Instance#

By default, Container uses the latest tag of the official EventSourcingDB Docker image. To change that, call the with_image_tag function:

container = (
  Container()
    .with_image_tag('1.0.0')
)

Similarly, you can configure the port to use and the API token. Call the with_port or the with_api_token function respectively:

container = (
  Container()
    .with_port(4000)
    .with_api_token('secret')
)

Added in 1.7If you want to sign events, call the with_signing_key function. This generates a new signing and verification key pair inside the container:

container = (
  Container()
    .with_signing_key()
)

You can retrieve the private key (for signing) and the public key (for verifying signatures) once the container has been started:

signing_key = container.get_signing_key()
verification_key = container.get_verification_key()

The signing_key is the private key EventSourcingDB signs events with. The verification_key can be passed to verify_signature when verifying events read from the database.

Configuring the Client Manually#

In case you need to set up the client yourself, use the following functions to get details on the container:

get_host() returns the host name
get_mapped_port() returns the port
get_base_url() returns the full URL of the container
get_api_token() returns the API token

Further Resources#

Repository: thenativeweb/eventsourcingdb-client-python
Registry: PyPI
License (MIT): License file
Issue tracker: GitHub Issues

Python SDK

Installation#

Capabilities#

Getting Started#

Configuring the Stall TimeoutAdded in 1.11#

Writing Events#

Using the isSubjectPristine precondition#

Using the isSubjectPopulated preconditionAdded in 1.8Requires EventSourcingDB 1.2#

Using the isSubjectOnEventId precondition#

Using the isEventQlQueryTrue preconditionChanged in 1.5#

Reading Events#

Reading From Subjects Recursively#

Reading in Anti-Chronological Order#

Specifying Bounds#

Starting From the Latest Event of a Given Type#

Aborting Reading#

Running EventQL Queries#

Aborting a Query#

Converting Events to pandas DataFrameAdded in 1.9#

Flattening the Data FieldAdded in 1.9#

Observing Events#

Observing From Subjects Recursively#

Specifying Bounds#

Starting From the Latest Event of a Given Type#

Detecting a Stalled ConnectionChanged in 1.11Changed in 2.0#

Aborting Observing#

Registering an Event Schema#

Listing Subjects#

Aborting Listing#

Listing Event Types#

Aborting Listing#

Listing a Specific Event TypeAdded in 1.5#

Verifying an Event's HashChanged in 1.11#

Verifying an Event's SignatureAdded in 1.7#

Using Testcontainers#

Configuring the Container Instance#

Configuring the Client Manually#

Further Resources#

Using the `isSubjectPristine` precondition#

Using the `isSubjectPopulated` preconditionAdded in 1.8Requires EventSourcingDB 1.2#

Using the `isSubjectOnEventId` precondition#

Using the `isEventQlQueryTrue` preconditionChanged in 1.5#