Viewing: docs.helix-db.com

Build HelixDB requests with idiomatic synchronous and asynchronous Python clients

Guide

The Python SDK is published on PyPI as helix-db 0.3.4 and imported as helixdb. Its builder methods use snake_case and produce the same operation-tree AST as the other HelixDB v3 SDKs.

helix-db includes the synchronous and asynchronous server clients. Install helix-db-embedded as well only when the application runs HelixDB in process.

Install

python -m pip install helix-db==0.3.4

Define a query

from helixdb import Predicate, define_params, g, param, read_batch

find_users_params = define_params({
    "tenant_id": param.string(),
    "limit": param.i64(),
})

def find_users():
    return (
        read_batch()
        .var_as(
            "users",
            g()
            .n_with_label("User")
            .where(Predicate.eq("tenantId", find_users_params.tenant_id))
            .limit(find_users_params.limit)
            .value_map(["$id", "name", "tenantId"]),
        )
        .returning(["users"])
    )

Execute

from helixdb import Client

request = find_users().to_query_request(
    find_users_params,
    {"tenant_id": "acme", "limit": 25},
    query_name="find_users",
)
result = Client("http://localhost:6969").query(request)

The server clients have no native runtime dependency.

Async execution

Reuse one async client to reuse its HTTP connection pool. HTTP requests have no timeout unless you explicitly configure one.

import asyncio

import httpx

from helixdb import AsyncClient

async def execute_queries():
    limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
    async with AsyncClient(timeout=10.0, limits=limits) as client:
        return await asyncio.gather(
            client.query(request),
            client.execute(request, writer_only=True, timeout=2.0),
        )

results = asyncio.run(execute_queries())

Cancellation propagates as asyncio.CancelledError, closes the response stream, and leaves the client reusable. The client owns and closes an injected HTTPX async transport. await client.close() is idempotent.

Embedded runtime

Install the SDK and embedded runtime with python -m pip install helix-db helix-db-embedded. See Embedded database for storage, cache, and handle configuration.

import asyncio

from helixdb import AsyncClient, Disk, InMemory

async def query_embedded():
    writer = await AsyncClient.embedded(InMemory("app"))
    async with writer:
        memory_result = await writer.query(request)

    reader = await AsyncClient.embedded_reader(Disk("./data", "seeded-app"))
    async with reader:
        disk_checkpoint = await reader.query(request)

    return memory_result, disk_checkpoint

results = asyncio.run(query_embedded())

Async embedded queries await native UniFFI operations directly. Use asyncio.timeout(...) for cancellation boundaries. Async native graph loading is not part of this API; use the synchronous Client.graph(...) API.

Verify

PYTHONDONTWRITEBYTECODE=1 \
PYTHONPATH=sdks/python/src \
python -m unittest discover sdks/python/tests

Start Here

Core Concepts

Query Guides

On this page

Python SDK

Install

Define a query

Execute

Async execution

Embedded runtime

Verify

Next steps

Query tutorial

Embedded database