Python SDK

Python SDK for building EvidentSource applications and server-side WASM components.

Status: Beta

SDK Components

Component	Status	Description
evidentsource-core	Stable	Core library: events, selectors, constraints, identifiers
evidentsource-client	Stable	Client library: gRPC connectivity to EvidentSource servers
evidentsource-functions	Beta	Server Functions library: WASM components for state changes and state views

Requirements

Python 3.12 or later
For WASM components: componentize-py

Installation

cd python

# Install all packages in development mode
pip install -e packages/evidentsource-core
pip install -e packages/evidentsource-client
pip install -e packages/evidentsource-functions

For WASM Development

# Install componentize-py
pip install componentize-py

# Build a state change
componentize-py -d ./interface/decider.wit -w state-change \
    componentize -o my_state_change.wasm my_state_change

Quick Start

Client Usage

import asyncio
from evidentsource_client import EvidentSource, DatabaseName

async def main():
    # Connect to the server
    es = await EvidentSource.connect("http://localhost:50051")

    # Connect to a database
    db_name = DatabaseName("my-database")
    conn = await es.connect_database(db_name)

    # Get the latest database state
    db = await conn.latest_database()
    print(f"Database: {db.name}")
    print(f"Revision: {db.revision}")

asyncio.run(main())

Server Functions Usage

State View

from dataclasses import dataclass
from evidentsource_functions import StateViewBase, Event
import json

@dataclass
class AccountSummary(StateViewBase):
    balance: float = 0.0

    def evolve_event(self, event: Event) -> None:
        if event.event_type == "account.credited":
            data = json.loads(event.data_as_string() or "{}")
            self.balance += data.get("amount", 0)

    def to_dict(self) -> dict:
        return {"balance": self.balance}

    @classmethod
    def from_dict(cls, data: dict) -> "AccountSummary":
        return cls(balance=data.get("balance", 0.0))

State Change

from dataclasses import dataclass
from evidentsource_functions import StateChangeError, DatabaseAccess, StateChangeMetadata
from evidentsource_core import ProspectiveEvent, StringEventData
import json
import uuid

@dataclass
class DepositCommand:
    account_id: str
    amount: float

def decide(
    db: DatabaseAccess,
    command: DepositCommand,
    metadata: StateChangeMetadata,
) -> tuple[list[ProspectiveEvent], list]:
    if command.amount <= 0:
        raise StateChangeError.validation("Amount must be positive")

    event = ProspectiveEvent(
        id=str(uuid.uuid4()),
        stream="accounts",
        event_type="account.credited",
        subject=command.account_id,
        data=StringEventData(json.dumps({
            "account_id": command.account_id,
            "amount": command.amount,
        })),
        datacontenttype="application/json",
    )

    return ([event], [])

Examples

Example	Description
TodoMVC	HTMX-based todo app with CRUD operations
Savings Account	Banking example with deposits, withdrawals, interest calculations

Running an Example

cd examples/todomvc  # or examples/savings-account

# Install dependencies
pip install -e .

# Build WASM components
./build_all.sh

# Install to EvidentSource server
./install.sh

# Run the server
python -m server.main

Testing

from evidentsource_functions import MockDatabase, TestEventBuilder

# Create a mock database
db = MockDatabase("test-db").with_revision(100)

# Insert state view data
db.insert_state_view_with_params(
    "account-summary", 1,
    {"account_id": "acct-123"},
    {"balance": 500.0, "status": "OPEN"},
    50
)

# Query state views
result = db.view_state("account-summary", 1, {"account_id": "acct-123"})
assert result.data["balance"] == 500.0

# Build test events
event = (
    TestEventBuilder("account.credited")
    .stream("accounts")
    .subject("acct-123")
    .data({"amount": 100.0})
    .build()
)

Project Structure

python/
├── packages/
│   ├── evidentsource-core/      # Foundation types
│   ├── evidentsource-client/    # gRPC client
│   └── evidentsource-functions/ # Server Functions library
└── examples/
    ├── todomvc/                 # HTMX todo app
    └── savings-account/         # Banking example

Development

# Run all package tests
cd packages/evidentsource-core && pytest
cd packages/evidentsource-functions && pytest

# Run example tests
cd examples/savings-account && pytest

# Type checking
pip install mypy
mypy packages/evidentsource-core/src
mypy packages/evidentsource-functions/src

API Reference

API documentation will be available on PyPI when packages are published.

Next Steps

Getting Started - Build your first EvidentSource application
State Changes - Learn about command handling
State Views - Learn about view materialization