umadb 0.1.5

Python client for UmaDB event store

UmaDB Python Client

A Python client for UmaDB event store built using Rust bindings via PyO3 and Maturin.

Installation

From PyPI (once published)

pip install umadb

From Source

# Install maturin
pip install maturin

# Build and install in development mode
cd crates/python
maturin develop

# Or build a wheel
maturin build --release

Usage

Basic Example

from umadb import Client, Event

# Connect to UmaDB server
client = Client("http://localhost:50051")

# Create and append events
event = Event(
    event_type="UserCreated",
    data=b"user data",
    tags=["user", "creation"],
)
position = client.append([event])
print(f"Event appended at position: {position}")

# Read events
events = client.read()
for seq_event in events:
    print(f"Position {seq_event.position}: {seq_event.event.event_type}")

# Get current head position
head = client.head()
print(f"Current head: {head}")

Using Queries

from umadb import Client, Query, QueryItem

client = Client("http://localhost:50051")

# Create a query to filter events
query_item = QueryItem(
    types=["UserCreated", "UserUpdated"],
    tags=["user"]
)
query = Query(items=[query_item])

# Read filtered events
events = client.read(query=query)
for seq_event in events:
    print(f"Position {seq_event.position}: {seq_event.event.event_type}")

Using Append Conditions

from umadb import Client, Event, Query, QueryItem, AppendCondition

client = Client("http://localhost:50051")

# Create a condition that prevents duplicate events
fail_query_item = QueryItem(types=["UserCreated"], tags=["user:123"])
fail_query = Query(items=[fail_query_item])
condition = AppendCondition(fail_if_events_match=fail_query)

# This will fail if matching events exist
event = Event(
    event_type="UserCreated",
    data=b"user data",
    tags=["user:123"],
)
try:
    position = client.append([event], condition=condition)
except ValueError as e:
    print(f"Append failed: {e}")

TLS/SSL Connection

from umadb import Client

# Connect with TLS using CA certificate
client = Client(
    url="https://secure-server:50051",
    ca_path="/path/to/ca.pem"
)

Reading with Options

from umadb import Client

client = Client("http://localhost:50051", batch_size=100)

# Read backwards from position
events = client.read(start=100, backwards=True, limit=10)

# Subscribe to new events (streaming)
events = client.read(subscribe=True)

API Reference

Client

Client(url: str, ca_path: str | None = None, batch_size: int | None = None)

Creates a new UmaDB client connection.

Methods:

• read(query=None, start=None, backwards=False, limit=None, subscribe=False): Read events from the store
• head(): Get the current head position (returns int | None)
• append(events, condition=None): Append events to the store (returns position as int)

Event

Event(event_type: str, data: bytes, tags: list[str] | None = None, uuid: str | None = None)

Represents an event in the event store.

Properties:

• event_type: Type of the event (string)
• data: Binary data (bytes)
• tags: List of tags (list of strings)
• uuid: Optional UUID (string)

SequencedEvent

Represents an event with its position in the sequence.

Properties:

• event: The Event object
• position: Position in the sequence (int)

Query

Query(items: list[QueryItem] | None = None)

A query for filtering events.

QueryItem

QueryItem(types: list[str] | None = None, tags: list[str] | None = None)

A query item specifying event types and tags to match.

AppendCondition

AppendCondition(fail_if_events_match: Query, after: int | None = None)

Condition for conditional appends.

Development

Building

# Install development dependencies
pip install maturin

# Build in development mode
maturin develop

# Run tests
pytest

Testing

The Python bindings can be tested by running a UmaDB server and executing Python code against it.

License

Licensed under either of:

• Apache License, Version 2.0 (LICENSE-APACHE)
• MIT license (LICENSE-MIT)

at your option.