FerricStore Python SDK

Python SDK for FerricStore and FerricFlow.

Status: public alpha 0.11.8. APIs may change before 1.0, but the SDK is tested against command construction, queue/workflow handlers, leases, retries, history, indexed attributes, named values, idempotent create, worker loops, async flows, and local FerricStore integration scenarios.

Python SDK 0.11.8 requires FerricStore 0.11.4 or newer. With FerricStore 0.11.10 it negotiates compact Stream mode 34 for homogeneous auto-ID XADD batches, compact Pub/Sub mode 35 for homogeneous PUBLISH pipelines, and the pubsub_batch_v1 event codec for compatible subscriptions. Native wire protocol v1 and generic compatibility paths remain.

FerricFlow keeps each workflow or job's state and history in one durable place. It is an explicit durable state pipeline, not a hidden deterministic replay engine:

create -> claim -> handler -> transition/complete/retry/fail

Handlers should be idempotent because work can be retried after lease expiry, worker crash, or explicit retry.

Durability is the default contract. A workflow command returns success only after the state change is accepted through FerricStore's quorum path and written to disk.

First 10 minutes

1. Install

pip install ferricstore

For local development from this repo:

python3 -m venv .venv
. .venv/bin/activate
pip install -e ".[dev]"

2. Start FerricStore

Use a local FerricStore server with the FerricStore protocol listener enabled:

ferricstore start

If you are running from the FerricStore source repo, use that repo's documented server command. The SDK examples assume:

ferric://127.0.0.1:6388

The same clients can send the same commands through a FerricStore HTTP endpoint:

from ferricstore import FlowClient

client = FlowClient.from_url(
    "https://ferricstore.example.com",
    username="platform_worker",
    password="service-secret",
)
client.kv_set("key", "value")
assert client.kv_get("key") == b"value"

The path is Python SDK -> HTTP(S) endpoint -> FerricStore. The endpoint can be the in-process ferricstore-http server or a compatible gateway. Use http:// / https:// for HTTP and ferric:// / ferrics:// for a direct native connection. HTTP is a deliberate request/response transport: ordinary commands and pipelines use the same client API, while live or connection-affine sessions such as Pub/Sub, WATCH, and transactions remain native-only. The current JSON envelope accepts JSON-compatible values; byte arguments use an automatic tagged Base64 envelope when arbitrary bytes or maps must cross JSON. See Command executors for the complete boundary.

For HTTP ACL authentication, the SDK sends the username and password using standard HTTP Basic authentication. This mode requires https://. Static bearer_token authentication remains available for compatible endpoint deployments. The SDK keeps a bounded pool of HTTP/TLS connections to the endpoint, so successive commands avoid repeated TCP and TLS handshakes. For multiplexing through an HTTP/2-capable TLS gateway, install ferricstore[http2] and pass http2=True; use max_connections to cap physical sockets and max_concurrent_requests to cap in-flight commands independently. Binary-heavy callers can install ferricstore[compact] and pass compact=True to use the endpoint's MessagePack envelope without changing command methods.

3. Query durable runs

Use parameterized FQL for bounded, partition-scoped reads. Cursors are opaque and must be reused with the same query and parameters.

from ferricstore import FlowClient

client = FlowClient.from_url("ferric://127.0.0.1:6388")
query = """
FROM runs
WHERE partition_key = @partition AND type = @type AND state = @state
ORDER BY updated_at_ms ASC
LIMIT 25
RETURN RECORDS
"""
params = {"partition": "partition-a", "type": "invoice", "state": "queued"}

result = client.query(query, params)
plan = client.explain(query, params)
indexes = client.query_indexes()

query_indexes() exposes each generation's bounded covering_fields and opaque format codec identities alongside build, validation, retirement, and statistics status. format.counter is None when an index has no exact-count prefix. A format change means the derived query projection must be rebuilt; authoritative Flow state is not rewritten.

For bounded request execution, pass deadline_ms as an absolute Unix timestamp in milliseconds to query, explain, or explain_analyze.

For reusable, composable queries, use the immutable query builder. It compiles to the same parameterized FQL1 and can be passed directly to query or explain:

from ferricstore import FlowFields, FlowQuery, flow_param

queued_invoices = (
    FlowQuery.runs()
    .where(
        FlowFields.partition_key.eq(flow_param("partition")),
        FlowFields.type.eq("invoice"),
        FlowFields.state.eq("queued"),
    )
    .order_by(FlowFields.updated_at_ms.desc())
    .limit(25)
    .return_records(FlowFields.run_id, FlowFields.state, FlowFields.updated_at_ms)
    .bind(partition="partition-a")
)

result = client.query(queued_invoices)

4. Create a durable queue item

from ferricstore import FlowClient, QueueClient

client = QueueClient.from_url("ferric://127.0.0.1:6388")
emails = client.queue(type="email")

emails.enqueue("email-1", payload=b"welcome:user-1", idempotent=True)

Use attributes for small indexed metadata you want to filter/count later:

emails.enqueue(
    "email-2",
    payload=b"welcome:user-2",
    partition_key="account-a",
    attributes={"account": "acme", "campaign": "summer"},
    idempotent=True,
)

flow = FlowClient.from_url("ferric://127.0.0.1:6388")
records = flow.list(
    "email",
    partition_key="account-a",
    attributes={"account": "acme"},
)
stats = flow.stats(
    "email",
    partition_key="account-a",
    attributes={"account": "acme"},
)

Attributes are not payload bytes. Use named values/value refs for large data.

FIFO Flow state policy is opt-in per state:

from ferricstore import FlowStatePolicy

flow.install_policy("email", states={"queued": FlowStatePolicy.fifo()})
emails.enqueue("email-3", payload=b"welcome", partition_key="account-a:email")

FIFO states require a partition_key; priority is for parallel states.

Policy writes are deep patches by default and return a typed snapshot with a monotonic generation. Use compare-and-swap when coordinating policy writers:

from ferricstore import StalePolicyGenerationError

snapshot = flow.policy_get("email")
try:
    snapshot = flow.install_policy(
        "email",
        expected_generation=snapshot.generation,
        states={"queued": FlowStatePolicy.fifo()},
    )
except StalePolicyGenerationError:
    snapshot = flow.policy_get("email")

Workflow install_policy() calls default to full replacement because the workflow definition is the source of truth. Pass replace=False to request a patch explicitly.

4. Run a queue worker

from ferricstore import QueueClient

client = QueueClient.from_url("ferric://127.0.0.1:6388")
emails = client.queue(type="email")


def send_email(job):
    print(f"send {job.id}: {job.payload!r}")
    return b"sent"


emails.worker(concurrency=10, batch_size=100).run(send_email)

If the handler raises, the default worker policy is retry.

5. Create a workflow/state machine

Use workflows when one durable flow moves through named states.

from ferricstore import WorkflowClient, complete, transition

client = WorkflowClient.from_url("ferric://127.0.0.1:6388")
order = client.workflow(
    type="order",
    initial_state="created",
    partition_by=("account_id", "order_id"),
)


@order.state("created")
def created(job):
    charge_card(job.payload)
    return transition("charged")


@order.state("charged")
def charged(job):
    send_receipt(job.id)
    return complete(result=b"ok")


order.start(
    "order-1",
    account_id="account-a",
    order_id="order-1",
    payload=b"order payload",
    idempotent=True,
)

order.worker(states=["created", "charged"], concurrency=10, batch_size=100).run()

Derived workflow partitions use collision-free fpk:<byte-length>:<value> encoding in 0.6.0. Drain flows created with the old colon-joined derived keys before upgrading. Explicit partition_key values are unchanged.

6. Store and fetch named values

Use named values when different states need different pieces of data. Values are stored as FerricFlow value refs and are only hydrated when requested.

emails.enqueue(
    "email-2",
    payload=b"small routing bytes",
    values={
        "template": b"welcome template bytes",
        "profile": b"user profile snapshot",
    },
    idempotent=True,
)

emails.worker(claim_values=["template"]).run(send_email)

Fetch one or many values directly when needed:

profile = client.value_get(owner_flow_id="email-2", name="profile")
values = client.value_mget(
    owner_flow_id="email-2",
    names=["template", "profile"],
)

Use ValueConfig or value_max_bytes in production to cap large value reads.

7. Inspect history

record = emails.get("email-1")
history = emails.history("email-1")

print(record)
for event in history:
    print(event)

History is for debugging and audit. Handlers should use claimed job data and requested values, not history replay.

8. Common errors

What you use

• QueueClient / AsyncQueueClient for durable queues.
• WorkflowClient / AsyncWorkflowClient for explicit durable state machines.
• FlowClient / AsyncFlowClient for advanced command-level control.
• ScheduleRecord, ScheduleFireResult, ScheduleFireDueResult, EffectResult, ApprovalResult, CircuitBreakerStatus, BudgetResult, and GovernanceOverview for typed admin/governance responses with dict fallback.
• RetryPolicy, WorkerConfig, ValueConfig, and ExceptionPolicy for runtime defaults.
• RawCodec by default, JsonCodec when you want JSON payloads.
• client.command(...) as the FerricStore low-level command escape hatch.

Async quickstart

import asyncio

from ferricstore import AsyncQueueClient


async def main():
    client = AsyncQueueClient.from_url("ferric://127.0.0.1:6388")
    emails = client.queue(type="email")

    async def handler(job):
        await send_email_async(job.payload)

    await emails.worker(concurrency=100, batch_size=500).run(handler)


asyncio.run(main())

Production shape

Use one process/service to create work and a separate long-lived worker service to claim and complete work.

web/serverless producer -> FerricStore -> worker service

Before production, configure timeouts, lease duration, backpressure behavior, graceful shutdown, and value hydration caps. The ferric:// transport defaults to one multiplexed connection with 8 request lanes; only raise connection or lane counts after profiling shows client-side saturation.

Docs

• Documentation index
• Quickstart
• SDK guide
• Configuration
• Production readiness
• Data in workflows
• Worker runtime
• Async APIs
• Use cases
• Testing
• Troubleshooting

Examples

• examples/order_workflow.py: two-state workflow.
• examples/queue_worker.py: queue producer and worker.
• examples/async_queue_worker.py: async queue producer and worker.
• examples/state_machine_workflow.py: explicit workflow runner.
• examples/protocol_commands.py: FerricStore command helpers.
• examples/protocol_kv_benchmark.py: protocol SET/GET benchmark.
• examples/protocol_dbos_benchmark.py: protocol DBOS-style queued workflow benchmark.
• examples/dbos_style_benchmark.py: DBOS-style throughput benchmark.

Contributing

See CONTRIBUTING.md, CODE_OF_CONDUCT.md, SECURITY.md, and RELEASE.md.