Skip to content
EvidentSource EvidentSource

Getting Started with Python

This guide walks you through building a complete EvidentSource application using the Python SDK.

Prerequisites

Section titled “Prerequisites”
  • Python 3.12 or later
  • componentize-py (for WASM components)
  • An EvidentSource server instance (local or cloud)

Quick Start with Templates

Section titled “Quick Start with Templates”

The fastest way to get started is using the project template:

Terminal window
# Clone the SDKs repository
git clone https://github.com/evidentsystems/evidentsource-sdks
cd evidentsource-sdks/python/packages/evidentsource-functions/templates


# Create a new project
./create-project.sh my_app my_database open_account active_accounts

This creates a complete project with:

  • domain/ - Shared types for events and commands
  • state_changes/ - Example state change component
  • state_views/ - Example state view component
  • build_all.sh - Build all WASM components
  • install.sh - Deploy components to EvidentSource
  • CLAUDE.md - AI assistant context for development
Terminal window
cd my_app
python -m venv .venv
source .venv/bin/activate
pip install -e .
./build_all.sh        # Build WASM components
./install.sh --all    # Deploy to server

See the Template Guide for full documentation.

Manual Project Setup

Section titled “Manual Project Setup”

If you prefer to set up manually or need a client-only application:

1. Clone the SDK Repository

Section titled “1. Clone the SDK Repository”
Terminal window
git clone https://github.com/evidentsystems/evidentsource-sdks.git
cd evidentsource-sdks/python

2. Install Dependencies

Section titled “2. Install Dependencies”
Terminal window
# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate


# Install SDK packages
pip install -e packages/evidentsource-core
pip install -e packages/evidentsource-client
pip install -e packages/evidentsource-functions

3. Create Project Structure

Section titled “3. Create Project Structure”
Terminal window
mkdir -p my-banking-app
cd my-banking-app


mkdir -p domain
mkdir -p state_changes/open_account
mkdir -p state_views/account_summary
mkdir -p server

Connecting to EvidentSource

Section titled “Connecting to EvidentSource”

Create server/main.py:

import asyncio
from evidentsource_client import EvidentSource, DatabaseName, StateChangeName, StateViewName


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

Authentication

Section titled “Authentication”

For production deployments, you’ll need to authenticate with a JWT token. See Authentication for details on the trust model.

import os
from evidentsource_client import EvidentSource, BearerTokenCredentials


async def main():
    # Bearer token authentication (production)
    token = os.environ["EVS_TOKEN"]
    es = await EvidentSource.connect(
        "api.example.com:50051",
        BearerTokenCredentials(token)
    )


    # DevMode for local development (no auth required)
    es = await EvidentSource.connect("http://localhost:50051")


    try:
        # Connect to a database
        db_name = DatabaseName("banking")
        conn = await es.connect_database(db_name)


        print(f"Connected at revision {conn.revision}")


        # Execute a state change
        sc_name = StateChangeName("open-account")
        db = await conn.execute_state_change(
            sc_name,
            1,
            {
                "accountId": "acct-001",
                "customerName": "Alice Smith",
                "initialDeposit": 1000.0,
            }
        )


        print(f"New revision: {db.revision}")


        # Query a state view
        sv_name = StateViewName("account-summary")
        view = await db.view_state(sv_name, 1, {"account_id": "acct-001"})


        if view:
            print(f"Balance: ${view.data['balance']:.2f}")


    finally:
        await conn.close()
        await es.close()


if __name__ == "__main__":
    asyncio.run(main())

Building a State Change

Section titled “Building a State Change”

State changes handle commands and emit events.

Create domain/events.py:

from dataclasses import dataclass


@dataclass
class AccountOpened:
    account_id: str
    customer_name: str


@dataclass
class AccountCredited:
    amount: float
    description: str


@dataclass
class AccountDebited:
    amount: float
    description: str

Create state_changes/open_account/main.py:

from dataclasses import dataclass
import json
import uuid


from evidentsource_functions import (
    StateChangeAdapter,
    StateChangeError,
    DatabaseAccess,
    StateChangeMetadata,
)
from evidentsource_core import ProspectiveEvent, StringEventData


@dataclass
class OpenAccountCommand:
    account_id: str
    customer_name: str
    initial_deposit: float


class OpenAccountStateChange(StateChangeAdapter[OpenAccountCommand]):
    def parse_command(self, body: bytes, content_type: str) -> OpenAccountCommand:
        data = json.loads(body)
        return OpenAccountCommand(
            account_id=data["accountId"],
            customer_name=data["customerName"],
            initial_deposit=data.get("initialDeposit", 0.0),
        )


    def decide(
        self,
        db: DatabaseAccess,
        cmd: OpenAccountCommand,
        metadata: StateChangeMetadata,
    ) -> tuple[list[ProspectiveEvent], list]:
        # Validate
        if not cmd.customer_name:
            raise StateChangeError.validation("Customer name is required")


        if cmd.initial_deposit < 0:
            raise StateChangeError.validation("Initial deposit cannot be negative")


        # Check if account exists
        existing = db.view_state("account-summary", 1, {"account_id": cmd.account_id})
        if existing is not None:
            raise StateChangeError.conflict("Account already exists")


        events = []


        # AccountOpened event
        events.append(ProspectiveEvent(
            id=str(uuid.uuid4()),
            stream="accounts",
            event_type="com.banking.account.opened",
            subject=cmd.account_id,
            data=StringEventData(json.dumps({
                "accountId": cmd.account_id,
                "customerName": cmd.customer_name,
            })),
            datacontenttype="application/json",
        ))


        # Initial deposit event
        if cmd.initial_deposit > 0:
            events.append(ProspectiveEvent(
                id=str(uuid.uuid4()),
                stream="accounts",
                event_type="com.banking.account.credited",
                subject=cmd.account_id,
                data=StringEventData(json.dumps({
                    "amount": cmd.initial_deposit,
                    "description": "Initial deposit",
                })),
                datacontenttype="application/json",
            ))


        return (events, [])


# Adapter instance for WASM binding
adapter = OpenAccountStateChange()

Build the WASM component:

Terminal window
cd state_changes/open_account
componentize-py -d ../../interface/decider.wit -w state-change \
    componentize -o open_account.wasm main

Building a State View

Section titled “Building a State View”

State views materialize read models from events.

Create state_views/account_summary/main.py:

from dataclasses import dataclass, field
import json
from typing import Optional


from evidentsource_functions import StateViewBase, Event


@dataclass
class AccountSummary(StateViewBase):
    id: str = ""
    name: str = ""
    balance: float = 0.0
    is_open: bool = False


    def evolve_event(self, event: Event) -> None:
        event_type = event.event_type


        if event_type == "com.banking.account.opened":
            data = self._parse_data(event)
            if data:
                self.id = data.get("accountId", "")
                self.name = data.get("customerName", "")
                self.is_open = True


        elif event_type == "com.banking.account.credited":
            data = self._parse_data(event)
            if data:
                self.balance += data.get("amount", 0.0)


        elif event_type == "com.banking.account.debited":
            data = self._parse_data(event)
            if data:
                self.balance -= data.get("amount", 0.0)


        elif event_type == "com.banking.account.closed":
            self.is_open = False


    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "name": self.name,
            "balance": self.balance,
            "isOpen": self.is_open,
        }


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


    def _parse_data(self, event: Event) -> Optional[dict]:
        try:
            data_str = event.data_as_string()
            if data_str:
                return json.loads(data_str)
        except json.JSONDecodeError:
            pass
        return None


# View instance for WASM binding
view = AccountSummary()

Build the WASM component:

Terminal window
cd state_views/account_summary
componentize-py -d ../../interface/decider.wit -w state-view \
    componentize -o account_summary.wasm main

Deploying Components

Section titled “Deploying Components”

Deploy WASM components to the EvidentSource server:

Terminal window
# Create database
curl -X POST http://localhost:8080/api/v1/databases/banking


# Install state change
curl -X POST http://localhost:8080/api/v1/databases/banking/state-changes/open-account/versions/1 \
  --data-binary @state_changes/open_account/open_account.wasm \
  -H "Content-Type: application/wasm"


# Install state view
curl -X POST http://localhost:8080/api/v1/databases/banking/state-views/account-summary/versions/1 \
  --data-binary @state_views/account_summary/account_summary.wasm \
  -H "Content-Type: application/wasm"

Testing

Section titled “Testing”

Create tests/test_open_account.py:

import pytest
from evidentsource_functions import MockDatabase, StateChangeMetadata
from state_changes.open_account.main import OpenAccountStateChange, OpenAccountCommand


def test_open_account_success():
    db = MockDatabase("test-db").with_revision(100)
    state_change = OpenAccountStateChange()


    cmd = OpenAccountCommand(
        account_id="acct-001",
        customer_name="Alice",
        initial_deposit=100.0,
    )


    events, conditions = state_change.decide(db, cmd, StateChangeMetadata())


    assert len(events) == 2  # opened + credited


def test_open_account_negative_deposit():
    db = MockDatabase("test-db")
    state_change = OpenAccountStateChange()


    cmd = OpenAccountCommand(
        account_id="acct-001",
        customer_name="Alice",
        initial_deposit=-50.0,
    )


    from evidentsource_functions import StateChangeError
    with pytest.raises(StateChangeError):
        state_change.decide(db, cmd, StateChangeMetadata())


def test_open_account_already_exists():
    db = MockDatabase("test-db")
    db.insert_state_view_with_params(
        "account-summary", 1,
        {"account_id": "acct-001"},
        {"id": "acct-001"},
        50
    )


    state_change = OpenAccountStateChange()
    cmd = OpenAccountCommand(
        account_id="acct-001",
        customer_name="Alice",
        initial_deposit=100.0,
    )


    from evidentsource_functions import StateChangeError
    with pytest.raises(StateChangeError):
        state_change.decide(db, cmd, StateChangeMetadata())

Run tests:

Terminal window
pytest tests/

Building a REST API

Section titled “Building a REST API”

Create server/api.py:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from evidentsource_client import EvidentSource, DatabaseName, StateChangeName, StateViewName


app = FastAPI()


es: EvidentSource = None


@app.on_event("startup")
async def startup():
    global es
    es = await EvidentSource.connect("http://localhost:50051")


@app.on_event("shutdown")
async def shutdown():
    await es.close()


class OpenAccountRequest(BaseModel):
    account_id: str
    customer_name: str
    initial_deposit: float = 0.0


class DepositRequest(BaseModel):
    amount: float
    description: str = ""


@app.get("/accounts/{account_id}")
async def get_account(account_id: str):
    conn = await es.connect_database(DatabaseName("banking"))
    try:
        db = await conn.latest_database()
        view = await db.view_state(
            StateViewName("account-summary"),
            1,
            {"account_id": account_id}
        )


        if not view:
            raise HTTPException(status_code=404, detail="Account not found")


        return view.data
    finally:
        await conn.close()


@app.post("/accounts")
async def open_account(request: OpenAccountRequest):
    conn = await es.connect_database(DatabaseName("banking"))
    try:
        db = await conn.execute_state_change(
            StateChangeName("open-account"),
            1,
            request.model_dump()
        )
        return {"revision": db.revision}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        await conn.close()


@app.post("/accounts/{account_id}/deposit")
async def deposit(account_id: str, request: DepositRequest):
    conn = await es.connect_database(DatabaseName("banking"))
    try:
        db = await conn.execute_state_change(
            StateChangeName("deposit"),
            1,
            {
                "accountId": account_id,
                **request.model_dump()
            }
        )
        return {"revision": db.revision}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        await conn.close()

Run the API:

Terminal window
pip install fastapi uvicorn
uvicorn server.api:app --reload

Next Steps

Section titled “Next Steps”