terminusdb-async

A small native async Python client for the TerminusDB HTTP API, built on HTTPX.

terminusdb-async is intended for asyncio applications that want direct, typed-ish access to TerminusDB document, schema, migration, and GraphQL endpoints without putting a synchronous SDK behind a thread pool.

Status: early-stage / pre-1.0. The public API may still evolve between 0.x releases.

Project status: this is a community package and is not an official TerminusDB client.

Installation

With uv:

uv add terminusdb-async

With pip:

python -m pip install terminusdb-async

Python 3.11+ is supported.

Quick start

import asyncio

from terminusdb_async import AsyncTerminusClient


async def main() -> None:
    async with AsyncTerminusClient(
        "http://localhost:6363",
        organization="admin",
        database="edtech",
        username="admin",
        password="root",
    ) as db:
        document = await db.get_document("Discipline/MATH-01")
        print(document)


asyncio.run(main())

The client owns its underlying httpx.AsyncClient by default and closes it when the async context manager exits.

Authentication

Basic authentication

client = AsyncTerminusClient(
    "http://localhost:6363",
    organization="admin",
    database="edtech",
    username="admin",
    password="root",
)

TerminusDB token

client = AsyncTerminusClient(
    "https://terminus.example.com",
    organization="acme",
    database="catalog",
    token="...",
)

Token authentication and basic authentication are mutually exclusive.

Document API

Read one document

doc = await db.get_document("Discipline/MATH-01")

Read documents

docs = await db.get_documents(
    document_type="Discipline",
    count=100,
)

You can also request explicit IDs:

docs = await db.get_documents(
    ids=["Discipline/MATH-01", "Discipline/PHYS-01"],
)

Insert documents

ids = await db.insert_documents(
    {
        "@type": "Discipline",
        "@id": "Discipline/MATH-01",
        "name": "Calculus",
    },
    author="application",
    message="Add Calculus",
)

A sequence of JSON-like mappings can be inserted in the same request.

Replace documents

await db.replace_documents(
    {
        "@type": "Discipline",
        "@id": "Discipline/MATH-01",
        "name": "Mathematical Analysis",
    },
    author="application",
    message="Rename discipline",
)

Delete documents

await db.delete_documents(
    "Discipline/MATH-01",
    author="application",
    message="Remove discipline",
)

Query documents

docs = await db.query_documents(
    {"name": "Calculus"},
    document_type="Discipline",
)

Schema API

Schema documents use the same Document API with graph_type="schema".

schema = await db.get_schema_documents()

await db.insert_schema_documents(
    [
        {
            "@type": "Class",
            "@id": "Discipline",
            "name": "xsd:string",
        }
    ],
    author="schema-bot",
    message="Add Discipline",
)

For initial schema loading, bootstrap_schema() performs a schema insert with full replacement enabled.

Migration API

The client exposes TerminusDB migration operations directly:

result = await db.migrate(
    [
        {
            "@type": "CreateClassProperty",
            "class": "Discipline",
            "property": "description",
            "type": {
                "@type": "Optional",
                "@class": "xsd:string",
            },
        }
    ],
    author="schema-bot",
    message="Add optional description",
    dry_run=True,
)

For automatic migration planning from Pydantic models, see terminusdb-migrations.

GraphQL

result = await db.graphql(
    """
    query {
      Discipline {
        _id
        name
      }
    }
    """
)

Variables can be passed with the variables= argument.

Server health and metadata

await db.ok()
await db.info()

Error handling

HTTP failures are converted to TerminusDBError:

from terminusdb_async import TerminusDBError

try:
    await db.get_document("Discipline/missing")
except TerminusDBError as exc:
    print(exc.status_code)
    print(exc.body)

The exception keeps the HTTP status code and the decoded error response when available.

Using an existing HTTPX client

You can inject an existing httpx.AsyncClient when connection pooling, transport customization, tracing, or application-wide lifecycle management is handled elsewhere:

import httpx

from terminusdb_async import AsyncTerminusClient

http = httpx.AsyncClient(base_url="http://localhost:6363")

db = AsyncTerminusClient(
    "http://localhost:6363",
    organization="admin",
    database="edtech",
    http_client=http,
)

An injected client is not closed by AsyncTerminusClient.

Resource paths and branches

The default branch is main. Requests target:

<organization>/<database>/local/branch/<branch>

Choose another TerminusDB branch explicitly:

db = AsyncTerminusClient(
    "http://localhost:6363",
    organization="admin",
    database="edtech",
    branch="feature-schema",
)

Scope

The package intentionally stays small. It currently focuses on:

• async HTTP transport;
• document CRUD and document queries;
• schema document access;
• native migration operations;
• GraphQL requests;
• Basic and TerminusDB token authentication.

It is not an ORM and it does not try to reproduce the full official TerminusDB Python SDK.

For model/schema integration, use terminusdb-pydantic.

Package family

All packages are pre-1.0 and are versioned independently.

Development

The project uses uv for dependency management, environments, building, and publishing.

uv sync --group dev
uv run pytest -q
uv build --no-sources

The test matrix covers Python 3.11, 3.12, and 3.13.

Development follows GitHub Flow on GitLab: short-lived branches are merged into main through Merge Requests. See CONTRIBUTING.md for the repository workflow.

Versioning

While the project is pre-1.0, consumers that want compatible updates should use a bounded dependency such as:

terminusdb-async = ">=0.1.1,<1.0.0"

Breaking changes may still occur in 0.x; review release notes before upgrading across minor versions.