vski-sdk

A fully-featured Python client for the VSKI backend — multi-tenant SQLite with collections, SQL, auth, realtime, and durable workflows. Async-first with a sync wrapper.

Install

pip install vski-sdk

Quickstart (async)

import asyncio
from vski import VskiClient

async def main():
    async with VskiClient("http://127.0.0.1:3000") as client:
        # Auth as admin (bootstrap the first superuser if needed)
        res = await client.admins.auth_with_password("admin@example.com", "password")
        client.set_token(res["token"])

        # Manage databases & collections
        await client.settings.databases.create({"name": "mydb"})
        await client.settings.collections.create({
            "name": "posts",
            "type": "base",
            "fields": [
                {"name": "title", "type": "text", "required": True},
                {"name": "views", "type": "number"},
            ],
        })

        # Records
        post = await client.collection("posts").create({"title": "Hello", "views": 0})
        page = await client.collection("posts").getList(1, 30, filter='title ~ "Hello"')

        # Raw SQL
        print(await client.sql.execute("SELECT count(*) FROM posts"))

asyncio.run(main())

Sync usage

from vski.sync import SyncVskiClient

with SyncVskiClient("http://127.0.0.1:3000") as client:
    res = client.admins.auth_with_password("admin@example.com", "password")
    client.set_token(res["token"])
    print(client.settings.databases.list())

Durable workflows

Author a workflow with the functional workflow() / step() API, then run a WorkflowWorker against it. Steps are idempotent and event-sourced, so runs survive worker restarts; sleep, wait_for_signal, parallel, retries, and rollback are all supported.

import asyncio
from vski import VskiClient, workflow, step, WorkflowWorker

@step("greet")
async def greet(name: str) -> list[str]:
    return [f"Hello, {name}"]

@workflow("hello").run
async def hello(ctx, name: str):
    return await greet(name)

async def main():
    async with VskiClient("http://127.0.0.1:3000") as client:
        res = await client.admins.auth_with_password("admin@example.com", "password")
        client.set_token(res["token"])

        worker = WorkflowWorker(client)
        task = asyncio.create_task(worker.start("hello"))
        await client.wait_for_workflow_ready()

        run = await client.workflow.trigger("hello", ["VSKI"])
        while True:
            run = await client.workflow.get_run(run["runId"])
            if run["status"] in ("completed", "failed"):
                break
            await asyncio.sleep(0.5)
        print(run["output"])  # ['Hello, VSKI']

        await worker.stop()
        await task

asyncio.run(main())

Features

• Databases & collections — create/list/delete databases, schema (fields, indexes, views, FTS, triggers), rules.
• Records — CRUD, search (FTS), views, bulk ops, file uploads, relation expand, auth-with-password.
• SQL & named queries — client.sql.execute(...), client.query.execute(...).
• Auth — admin/user password auth, registration, refresh, OAuth2, API keys, RBAC.
• Automation — cron jobs, webhooks + logs, gates (API proxy).
• Realtime — WebSocket collection subscriptions with consumer groups.
• Durable workflows — workflow() / step() authoring, WorkflowWorker, signals, sleeps, parallel, rollback, event-sourced recovery.
• Replication — list known replicas; sticky replica routing via X-Replica-Id.

Compatibility

Wire-compatible with the VSKI server (vski) and equivalent to the TypeScript @vski/sdk. License: VSKI License v1.0 — see LICENSE.

Testing

The tests/ directory is a full port of the TypeScript e2e suites (test/e2e/ and test/e2e-replica/) — 98 scenarios covering every feature, including the complete durable-workflow matrix (signals, sleeps, retries, parallel, saga rollback, circuit breakers, child workflows, recovery) and replication. The pytest harness builds/locates vski-prod, starts a fresh server per session, and tears it down (process-group managed).

# Run the core e2e suite (master on :3001)
pytest tests/e2e/

# Run the replication suite (master on :3001 + replica on :3002)
pytest tests/e2e_replica/

Set VSKI_BINARY, API_URL, or VSKI_REUSE_SERVER=1 to target a specific or already-running server.

Improvements over the TypeScript SDK

• Unified VskiError (status + parsed message) instead of per-namespace throw new Error(text()).
• Configurable request timeout + retries on transient errors (the TS SDK had none).
• Lazily-cached namespace singletons (TS rebuilt every namespace on each access).
• Realtime supports multiple subscriptions per collection (TS overwrote prior callbacks).
• Concurrency-safe workflow context via contextvars (TS fell back to a global in browsers).
• Fixed 500ms duration parsing (TS checked s before ms, misparsing it as 500s).
• Enforced worker concurrency cap (TS accepted but ignored it).