Skip to main content

gRPC client for EventStore DB

Project description

esdb-py

EventStoreDB Python gRPC client

NOTE: This project is still work in progress

Implemented parts

  • secure connection
  • basic auth
  • other connection options
    • multi-node gossip
    • keepalive
  • async client
  • streams
    • append
    • batch append
    • delete
    • read
    • tombstone
    • filtering
    • exception handling
  • subscriptions
  • users
  • tbd

Setting things up

  1. Install poetry
  2. Create virtualenv (i.e. using pyenv):
pyenv install 3.10.5
pyenv virtualenv 3.10.5 esdb-py
pyenv local esdb-py
  1. Install deps with poetry install
  2. Start eventstore in docker: make run-esdb
  3. Run the tests: pytest tests

Usage:

import datetime
import uuid

from esdb.client.client import ESClient

# For insecure connection without basic auth:
# client = ESClient("localhost:2113", tls=False)
with open("certs/ca/ca.crt", "rb") as fh:
  root_cert = fh.read()

client = ESClient("localhost:2111", root_certificates=root_cert, username="admin", password="changeit")

stream = f"test-{str(uuid.uuid4())}"

for i in range(10):
  append_result = client.streams.append(
    stream=stream,
    event_type="test_event",
    data={"i": i, "ts": datetime.datetime.utcnow().isoformat()},
  )

print("Forwards!")
for result in client.streams.read(stream=stream, count=10):
  print(result.data)

print("Backwards!")
for result in client.streams.read(stream=stream, count=10, backwards=True):
  print(result.data)

print("Forwards start from middle!")
for result in client.streams.read(stream=stream, count=10, revision=5):
  print(result.data)

print("Backwards start from middle!")
for result in client.streams.read(stream=stream, count=10, backwards=True, revision=5):
  print(result.data)

Async example:

import asyncio

from esdb.client.client import AsyncESClient


async def append():
  client = AsyncESClient("localhost:2113")
  result = await client.streams.append("stream", "type", {"x": 1})
  assert result.commit_position > 0
  async for event in client.streams.read("stream", count=10):
    print(event)


asyncio.run(append())

Subscriptions:

from esdb.client.client import ESClient

client = ESClient("localhost:2113", tls=False)
stream = "stream-name"
group = "group-name"

# emit some events to the same stream
for _ in range(10):
    client.streams.append(stream, "foobar", b"data")

# create a subscription
client.subscriptions.create_stream_subscription(stream=stream, group_name=group)

# Read from subscription
# This will block and wait for messages
subscription = client.subscriptions.subscribe_to_stream(stream, group)
for event in subscription:
    # ... do work with the event ...
    # ack the event
    subscription.ack([event])

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

esdb-0.1.0.tar.gz (62.6 kB view hashes)

Uploaded Source

Built Distribution

esdb-0.1.0-py3-none-any.whl (78.5 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page