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.

overwrite=True forwards the server's existing-ID insertion option. It does not remove old property values. Use replace_documents() to update an existing document.

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

Unit tests run without a server. To run the integration suite against an isolated local server:

docker compose -f compose.test.yml up -d
uv run pytest -q --run-integration
docker compose -f compose.test.yml down -v

To choose the other tested version, prefix the up command with TERMINUSDB_VERSION=v12.0.6. For an existing test server, set TERMINUSDB_TEST_URL, TERMINUSDB_TEST_USERNAME, TERMINUSDB_TEST_PASSWORD, and TERMINUSDB_TEST_ORGANIZATION as needed (defaults: localhost:6363, admin/root, organization admin). Each test creates a unique database and deletes it afterwards. The suite waits up to 120 seconds for readiness; once enabled, an unavailable server fails the tests rather than skipping them.

GitLab CI runs unit tests on Python 3.11, 3.12, and 3.13, plus the full integration suite for every combination of those Python versions and TerminusDB v12.0.6 / v12.0.7. The service uses Basic authentication. Integration coverage includes schema bootstrap/read/write, document CRUD, batch operations, filters, pagination, server errors, invalid authentication, migration dry-run/apply, GraphQL variables, branch isolation, and concurrent reads. JUnit reports are available in the pipeline. All tests must succeed before building and publishing a release. Other server versions are not currently verified by CI.

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.2,<1.0.0"

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

See CHANGELOG.md for release notes.