Advanced

Async Support

Use AsyncCollection and AsyncTimeSeriesCollection for non-blocking database operations in async applications.

CentauroDB provides async wrappers for both collection types. Every method uses asyncio.to_thread under the hood, so database operations don't block your event loop. This guide covers setup, usage patterns, and integration with async frameworks.

Setup

Create the engine with check_same_thread=False — async operations run in a background thread, so SQLite's default thread-safety check must be disabled:

from centaurodb import Engine, AsyncCollection, CentauroModel

engine = Engine("app.db", check_same_thread=False)
items = AsyncCollection(engine, "items")

If you forget check_same_thread=False, creating an async collection raises ThreadSafetyError. See the Engine API for all constructor parameters.

Async collections share the same Engine instance as their sync counterparts — no separate connection pool is needed.

Complete async workflow

All methods mirror their sync counterparts, but return awaitables:

class Task(CentauroModel):
    __centauro_name__ = "Task"
    title: str = ""
    status: str = "pending"
    priority: int = 0

async def main():
    # Write
    task = Task(title="Slay the Hydra", status="pending", priority=2)
    await items.write_object(task)

    # Batch write
    await items.write_objects([
        Task(title="Retrieve the Golden Fleece", priority=3),
        Task(title="Map the Labyrinth", priority=1),
    ])

    # Read
    loaded = await items.read_object_by_id(task.row.id, Task)
    print(loaded.title)  # "Slay the Hydra"

    # Query
    pending = await items.read_objects(
        Task.fields.status == "pending"
    )
    high_priority = await items.read_objects(
        Task.fields.priority > 1
    )

    # Latest
    latest = await items.read_latest_object(Task)

    # Update
    loaded.status = "done"
    await items.update_object(loaded)

    # Refresh
    await items.refresh(loaded)

    # Delete
    await items.delete_object(loaded)

    # Bulk delete
    count = await items.delete_objects(Task.fields.status == "done")

AsyncTimeSeriesCollection

from centaurodb import AsyncTimeSeriesCollection, CentauroModelSeries, CentauroValues

engine = Engine("metrics.db", check_same_thread=False)
metrics = AsyncTimeSeriesCollection(engine, "metrics")

class CPUMetric(CentauroModelSeries):
    __centauro_name__ = "CPUMetric"
    host: str = ""

async def ingest():
    metric = CPUMetric(
        host="pegasus-01",
        values=[
            CentauroValues(time="2025-06-01T08:00:00", value=45.2),
            CentauroValues(time="2025-06-01T08:01:00", value=62.1),
        ],
    )
    await metrics.write_object(metric)

    # Read with values
    loaded = await metrics.read_object_by_id(metric.row.id, CPUMetric)
    print(loaded.df)

    # Read metadata only
    light = await metrics.read_object_by_id(
        metric.row.id, CPUMetric, hydrate_values=False
    )

    # Update with append mode
    metric.values = [CentauroValues(time="2025-06-01T08:02:00", value=55.8)]
    await metrics.update_object(metric, values_mode="append")

Views and indexes (async)

View operations are also async:

await items.create_view("dashboard", Task)

# Indexes are managed on the sync base — no await needed
items.create_index("status")
items.create_index("priority")
await items.refresh_view("dashboard", Task)
views = await items.describe_views()

Relations (async)

Parent-child operations work the same way:

from centaurodb import parent_ref

class Project(CentauroModel):
    __centauro_name__ = "Project"
    name: str = ""

class Issue(CentauroModel):
    __centauro_name__ = "Issue"
    project: int | None = parent_ref(Project)
    title: str = ""

async def demo():
    coll = AsyncCollection(engine, "quests")

    project = Project(name="The Odyssey")
    await coll.write_object(project)

    issue = Issue(title="Chart a course past Scylla")
    await coll.write_child(issue, parent=project)

    issues = await coll.children_of(project, Issue)

Concurrent operations

Since async collections use to_thread, you can run multiple independent operations concurrently:

import asyncio

async def parallel_reads():
    sensor_a, sensor_b, latest = await asyncio.gather(
        metrics.read_object_by_id(1, CPUMetric),
        metrics.read_object_by_id(2, CPUMetric),
        metrics.read_latest_object(CPUMetric),
    )

For truly concurrent writes, make sure each coroutine operates on separate objects. CentauroDB serializes writes through the underlying database connection.

PostgreSQL

With PostgreSQL, check_same_thread is not needed — psycopg connections are thread-safe by default:

engine = Engine("postgresql://user:pass@localhost/mydb")
items = AsyncCollection(engine, "items")  # works without check_same_thread

See the PostgreSQL guide for connection setup and backend differences.

Edit this pageorReport an issue

Time-Series

Store and query time-series data with TimeSeriesCollection, Polars DataFrames, and flexible update modes.

PostgreSQL

Switch from SQLite to PostgreSQL — same API, server-grade backend.