Python SDK

python -m pip install helix-db

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"])
    )

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)

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())

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())

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