Python Examples

The Python examples use virtufin-api, which reaches every service through the gateway: state and pub/sub directly, WorkManager and WebSocketManager as client.gateway.workmanager.<Method>(...) and client.gateway.websocketmanager.<Method>(...).

Service SDK

Package	Registry	Install
virtufin-api	Gitea PyPI	pip install virtufin-api --index-url ...

Connecting: the context

Each example sets CONTEXT_NAME near its top, "demo" by default: a name in ~/.config/virtufin/api/contexts.toml (see Getting Started). python/virtufin_context.py reads the entry and builds the client:

from virtufin_context import api_client, describe, load_context

CONTEXT_NAME = "demo"  # or "local" for a deploy-local backend

context = load_context(CONTEXT_NAME)
print(describe(context))  # context demo: api.virtufin.com:443 tls=True key=set
api = api_client(context)  # ApiClient(host, grpc_port, tls, api_key)

Key Patterns

# State: the service must be one the gateway registers ("statestore" on demo and local).
await api.save_state_async("statestore", "my-key", json.dumps(data))
data = json.loads((await api.get_state_async("statestore", "my-key"))["value"])

# WebSocketManager through the gateway. On demo, always disconnect (in `finally`).
conn = await api.gateway.websocketmanager.Connect({"url": "wss://...", "auto_reconnect": True})
try:
    await api.gateway.websocketmanager.StartPublish({"id": conn["id"], "topic": "my-topic"})
finally:
    await api.gateway.websocketmanager.Disconnect({"id": conn["id"]})

# Pub/sub
async for event in api.pubsub.subscribe_async_iter(topic="my-topic"):
    payload = json.loads(event.data_as_text())

Dependencies

virtufin-api
python-dotenv
grpcio