Metadata-Version: 2.4
Name: terminusdb-async
Version: 0.1.2
Summary: Native async Python client for the TerminusDB HTTP API
Project-URL: Source, https://gitlab.com/ed-tech6370840/terminusdb-async
Project-URL: Issues, https://gitlab.com/ed-tech6370840/terminusdb-async/-/issues
Project-URL: PyPI, https://pypi.org/project/terminusdb-async/
Keywords: terminusdb,asyncio,httpx,graph-database,document-database
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: httpx<1,>=0.27

# terminusdb-async

[![PyPI](https://img.shields.io/pypi/v/terminusdb-async.svg)](https://pypi.org/project/terminusdb-async/)
[![Python](https://img.shields.io/pypi/pyversions/terminusdb-async.svg)](https://pypi.org/project/terminusdb-async/)

A small native async Python client for the TerminusDB HTTP API, built on
[HTTPX](https://www.python-httpx.org/).

`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:

```bash
uv add terminusdb-async
```

With pip:

```bash
python -m pip install terminusdb-async
```

Python 3.11+ is supported.

## Quick start

```python
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

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

### TerminusDB token

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

Token authentication and basic authentication are mutually exclusive.

## Document API

### Read one document

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

### Read documents

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

You can also request explicit IDs:

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

### Insert documents

```python
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

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

### Delete documents

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

### Query documents

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

## Schema API

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

```python
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:

```python
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`](https://pypi.org/project/terminusdb-migrations/).

## GraphQL

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

Variables can be passed with the `variables=` argument.

## Server health and metadata

```python
await db.ok()
await db.info()
```

## Error handling

HTTP failures are converted to `TerminusDBError`:

```python
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:

```python
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:

```text
<organization>/<database>/local/branch/<branch>
```

Choose another TerminusDB branch explicitly:

```python
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`](https://pypi.org/project/terminusdb-pydantic/).

## Package family

| Package | Purpose |
| --- | --- |
| [`terminusdb-async`](https://pypi.org/project/terminusdb-async/) | Async TerminusDB HTTP client |
| [`terminusdb-pydantic`](https://pypi.org/project/terminusdb-pydantic/) | Pydantic v2 ↔ JSON-LD and schema generation |
| [`terminusdb-migrations`](https://pypi.org/project/terminusdb-migrations/) | Schema diff and migration planning |

All packages are pre-1.0 and are versioned independently.

## Development

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

```bash
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:

```bash
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:

```toml
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](CHANGELOG.md) for release notes.
