API Reference

Collection

API reference for Collection and TimeSeriesCollection.

Collection

A named collection of objects stored in a {name}_objects table. Each object is a Pydantic model serialized as a JSON blob alongside standard metadata columns.

Constructor

from centaurodb import Collection, Engine

engine = Engine("my.db")
coll = Collection(engine, "monitoring")

Parameter	Type	Description
`engine`	`Engine`	The database engine
`name`	`str`	Collection name — lowercase alphanumeric with underscores, must start with a letter

The table monitoring_objects is created on first instantiation if it doesn't already exist. Creating a Collection with the same name twice is safe and idempotent.

`write_object`

Insert a new object into the collection. Sets obj.row and marks the object as persisted.

coll.write_object(obj)

Parameters

Parameter	Type	Default	Description
`obj`	`CentauroModel`	—	The model instance to persist
`duplicate`	`bool`	`False`	If `True`, insert a new row even if this object was already written

Raises

ObjectAlreadyPersistedError — if obj was already written and duplicate=False
UnlinkedParentError — if the object has any parent_ref fields that are still None
TypeError — if obj is not a CentauroModel instance, or if it is a CentauroModelSeries (use TimeSeriesCollection for those)

sensor = Sensor(unit="Celsius", location="pelion")
coll.write_object(sensor)

print(sensor.row.id)          # 1
print(sensor.row.write_time)  # datetime(...)

# Insert a fresh copy of the same data as a new row
coll.write_object(sensor, duplicate=True)
print(sensor.row.id)          # 2

`write_objects`

Insert multiple objects in a single atomic transaction. If any object fails validation, none are persisted and the transaction is rolled back.

coll.write_objects(objs)

Parameters

Parameter	Type	Default	Description
`objs`	`Sequence[CentauroModel]`	—	A sequence of model instances to persist
`duplicate`	`bool`	`False`	If `True`, force insert new rows even for already-written objects

Raises

ObjectAlreadyPersistedError — if any object was already written and duplicate=False
TypeError — if any element is not a CentauroModel instance, or is a CentauroModelSeries (use TimeSeriesCollection)

books = [Book(title="Dune"), Book(title="Foundation")]
coll.write_objects(books)
print(all(b.row.id for b in books))  # True — all persisted

`read_object_by_id`

Read a single object from the collection by its row id.

obj = coll.read_object_by_id(id, ModelClass)

Parameters

Parameter	Type	Default	Description
`id`	`int`	—	Row id to look up
`model_class`	`type[CentauroModel]`	—	The model class to deserialize into
`raw`	`bool`	`False`	If `True`, return the raw database row dict instead of a model instance

Returns — a deserialized model instance, or a dict-like row when raw=True.

Raises — ObjectNotFoundError if no row with that id exists.

sensor = coll.read_object_by_id(1, Sensor)
print(sensor.unit)          # "Celsius"
print(sensor.row.id)        # 1
print(sensor.row.name)      # "Sensor"

# Raw row — useful for debugging or direct SQL inspection
row = coll.read_object_by_id(1, Sensor, raw=True)
print(row["meta"])          # '{"unit":"Celsius","location":"pelion",...}'
print(row["write_time"])    # "2025-06-01T08:00:00+00:00"

`read_latest_object`

Read the most recently edited object of a given type. Objects are ordered by edit_time DESC, id DESC.

obj = coll.read_latest_object(ModelClass)

Parameters

Parameter	Type	Default	Description
`model_class`	`type[CentauroModel]`	—	The model class to query
`raw`	`bool`	`False`	If `True`, return the raw database row dict instead of a model instance

Returns — a deserialized model instance, a dict-like row when raw=True, or None if no objects of that type exist.

latest = coll.read_latest_object(Sensor)
if latest:
    print(latest.unit)
    print(latest.row.edit_time)

# Raw row
row = coll.read_latest_object(Sensor, raw=True)

`read_objects`

Read objects from the collection, optionally filtered by conditions. Accepts three calling styles:

Model class — return all objects of that type
Conditions — return objects matching filters (model class inferred)
FieldProxy (legacy) — Model.fields alone returns all objects

results = coll.read_objects(Sensor)
results = coll.read_objects(Sensor.fields.unit == "Celsius")

Parameters

Parameter	Type	Description
`*args`	`type[CentauroModel]` or `FieldProxy` or `Condition` or `OrGroup`	Model class or query conditions — see examples
`raw`	`bool`	If `True`, return raw rows instead of model instances
`max_field`	`Field`	Select rows matching the maximum value of this field
`min_field`	`Field`	Select rows matching the minimum value of this field
`limit`	`int \| None`	Maximum number of rows to return
`offset`	`int \| None`	Number of rows to skip — requires `limit`

Build conditions via Model.fields:

# All objects of a type (pass the class directly)
all_sensors = coll.read_objects(Sensor)

# Legacy style — also returns all objects of that type
all_sensors = coll.read_objects(Sensor.fields)

# Single equality condition
results = coll.read_objects(Sensor.fields.unit == "Celsius")

# Multiple AND conditions (pass as separate arguments)
results = coll.read_objects(
    Sensor.fields.unit == "Celsius",
    Sensor.fields.threshold > 50,
)

# OR conditions
results = coll.read_objects(
    (Sensor.fields.unit == "Celsius") | (Sensor.fields.unit == "Kelvin"),
)

# OR + AND
results = coll.read_objects(
    (Sensor.fields.unit == "Celsius") | (Sensor.fields.unit == "Kelvin"),
    Sensor.fields.threshold > 50,
)

# Aggregate: row(s) with the highest threshold
results = coll.read_objects(Sensor, max_field=Sensor.fields.threshold)

# Aggregate: earliest sensor by write_time
results = coll.read_objects(Sensor, min_field=Sensor.fields.write_time)

# Aggregate + filter: highest threshold among Celsius sensors
results = coll.read_objects(
    Sensor.fields.unit == "Celsius",
    max_field=Sensor.fields.threshold,
)

max_field / min_field only work with int, float, date, or datetime fields. They are mutually exclusive — pass only one at a time.

Paginate with limit / offset — combine with count_objects to drive a pager:

# First page of 20
page = coll.read_objects(Sensor, limit=20)

# Third page
page = coll.read_objects(Sensor, limit=20, offset=40)

# Total for the pager UI
total = coll.count_objects(Sensor)

offset without limit raises ValueError — pagination without an upper bound is rejected to avoid unbounded reads.

Available operators on field conditions

Operator	Example
`==`	`Sensor.fields.unit == "Celsius"`
`!=`	`Sensor.fields.unit != "Fahrenheit"`
`>` / `>=`	`Sensor.fields.threshold > 50`
`<` / `<=`	`Sensor.fields.threshold <= 100`
`.like(pattern)`	`Sensor.fields.location.like("lab%")`
`.is_in(values)`	`Sensor.fields.unit.is_in(["Celsius", "Kelvin"])`
`\|`	`(cond_a) \| (cond_b)`

Model.fields supports id, write_time, edit_time, and all user-defined fields. The name column is not queryable via Model.fields — results are already filtered to the model's __centauro_name__ automatically.

Raises

TypeError — if no arguments are given, or Model.fields is mixed with other conditions
ValueError — if conditions reference different model classes, or an unknown field name, or if limit/offset are negative, or if offset is passed without limit

`count_objects`

Count objects matching a query without materializing any rows. Accepts the same calling styles as read_objects; results are always filtered to the model's __centauro_name__.

total = coll.count_objects(Sensor)

Parameters

Parameter	Type	Description
`*args`	`type[CentauroModel]` or `FieldProxy` or `Condition` or `OrGroup`	Model class or query conditions — same as `read_objects`

Returns — int, the number of matching rows.

# Count all sensors
total = coll.count_objects(Sensor)

# Count with conditions
celsius = coll.count_objects(Sensor.fields.unit == "Celsius")

# Pairs with limit/offset for client-side paging
total = coll.count_objects(Sensor)
page = coll.read_objects(Sensor, limit=20, offset=40)

`aggregate`

Run an aggregate query over objects matching the same calling styles as read_objects.

total = coll.aggregate(Sensor, count=True)

Parameters

Parameter	Type	Default	Description
`*args`	`type[CentauroModel]` or `FieldProxy` or `Condition` or `OrGroup`	—	Model class or query conditions
`count`	`bool`	`False`	If `True`, return the matching row count
`sum`	`Field`	`None`	Reserved — raises `NotImplementedError`
`avg`	`Field`	`None`	Reserved — raises `NotImplementedError`
`group_by`	`Field`	`None`	Reserved — raises `NotImplementedError`

Returns — int when count=True. The return type is deliberately wide (int | float | dict | polars.DataFrame) so future aggregates won't change the signature.

Raises

ValueError — if no aggregate is requested
NotImplementedError — if sum, avg, or group_by are passed

As of 0.9.0 only count=True is implemented. sum, avg, and group_by are reserved keywords that raise NotImplementedError — the signature is locked now so code written today keeps working when they ship. For SQL aggregations today, create a view and use sql_select:

df = coll.sql_select(
    "SELECT unit, AVG(threshold) FROM monitoring_view_dashboard GROUP BY unit"
)

# Equivalent to count_objects(Sensor)
total = coll.aggregate(Sensor, count=True)

# Count with conditions
hot = coll.aggregate(Sensor.fields.threshold > 50, count=True)

`update_object`

Persist the current state of an already-written object. Updates edit_time; write_time is preserved.

coll.update_object(obj)

Parameters

Parameter	Type	Description
`obj`	`CentauroModel`	A previously persisted model instance

Raises

ObjectNotPersistedError — if write_object was never called on this instance
UnlinkedParentError — if any parent_ref fields are None
TypeError — if obj is not a CentauroModel instance

sensor = coll.read_object_by_id(1, Sensor)
sensor.unit = "Kelvin"
coll.update_object(sensor)

refreshed = coll.read_object_by_id(1, Sensor)
print(refreshed.unit)           # "Kelvin"
print(refreshed.row.edit_time)  # updated timestamp

`refresh`

Re-read the object's fields and row metadata from the database in place. Useful after another process may have updated the same row.

coll.refresh(obj)

Parameters

Parameter	Type	Description
`obj`	`CentauroModel`	A persisted model instance to update in place

Raises

ObjectNotPersistedError — if the object has never been written
ObjectNotFoundError — if the row no longer exists (e.g. deleted externally)

sensor = coll.read_object_by_id(1, Sensor)

# ... some other process updates row 1 ...

coll.refresh(sensor)
print(sensor.unit)  # reflects the latest database state

`delete_object`

Delete a single object from the collection. Resets the object's persistence state so it can be re-written if needed.

coll.delete_object(obj)

Parameters

Parameter	Type	Description
`obj`	`CentauroModel`	A persisted model instance to delete

Raises

ObjectNotPersistedError — if write_object was never called on this instance
ParentHasChildrenError — if the object is a parent with children linked via parent_ref(on_delete="restrict") (the default)
NotImplementedError — if children are linked via the reserved on_delete modes "cascade" or "set_null"
TypeError — if obj is not a CentauroModel instance

sensor = coll.read_object_by_id(1, Sensor)
coll.delete_object(sensor)

# Object is no longer persisted
sensor.row  # raises AttributeError

`delete_objects`

Delete objects from the collection, optionally filtered by conditions. Accepts the same calling styles as read_objects.

count = coll.delete_objects(Sensor)
count = coll.delete_objects(Sensor.fields.unit == "Celsius")

Parameters

Parameter	Type	Description
`*args`	`type[CentauroModel]` or `Condition` or `OrGroup`	Model class or query conditions — same as `read_objects`

Returns — int, the number of rows deleted.

Raises

TypeError — if no arguments are given or arguments are of incompatible types
ValueError — if conditions reference different model classes or an unknown field name
ParentHasChildrenError — if any matched object is a parent with children linked via parent_ref(on_delete="restrict") (the default)
NotImplementedError — if children are linked via the reserved on_delete modes "cascade" or "set_null"

# Delete all sensors
count = coll.delete_objects(Sensor)
print(count)  # 5

# Delete sensors matching a condition
count = coll.delete_objects(Sensor.fields.unit == "Fahrenheit")
print(count)  # 2

# Delete with OR conditions
count = coll.delete_objects(
    (Sensor.fields.unit == "Celsius") | (Sensor.fields.unit == "Kelvin"),
)

`write_child`

Link a child to its parent and write it in one step. Internally sets the matching parent_ref field to parent.row.id, then calls write_object(child).

coll.write_child(child, parent=parent)

Parameters

Parameter	Type	Default	Description
`child`	`CentauroModel`	—	The child object to link and persist
`parent`	`CentauroModel`	—	The parent object. Must be persisted
`duplicate`	`bool`	`False`	If `True`, force insert a new row even if the child was already written

Raises

TypeError — if the child has no parent_ref pointing to the parent's model class
ObjectNotPersistedError — if the parent has not been written
ObjectAlreadyPersistedError — if the child was already written and duplicate=False

portfolio = Portfolio(strategy="momentum")
coll.write_object(portfolio)

trade = Trade(symbol="AAPL", quantity=100)
coll.write_child(trade, parent=portfolio)

print(trade.portfolio)   # portfolio.row.id
print(trade.row.id)      # auto-assigned

`children_of`

Read all children linked to a parent object. Auto-discovers the foreign-key field by inspecting the child class for a parent_ref pointing to the parent's model class.

children = coll.children_of(parent, ChildClass)

Parameters

Parameter	Type	Default	Description
`parent`	`CentauroModel`	—	The parent object. Must be persisted
`child_class`	`type[CentauroModel]`	—	The child model class to query
`raw`	`bool`	`False`	If `True`, return raw database rows instead of model instances

Returns — a list of child model instances, or raw rows if raw=True.

Raises

TypeError — if the child class has no parent_ref pointing to the parent's model class
ObjectNotPersistedError — if the parent has not been written

portfolio = Portfolio(strategy="momentum")
coll.write_object(portfolio)

t1 = Trade(symbol="AAPL")
t2 = Trade(symbol="GOOG")
coll.write_child(t1, parent=portfolio)
coll.write_child(t2, parent=portfolio)

trades = coll.children_of(portfolio, Trade)
print(len(trades))  # 2

# Raw rows
rows = coll.children_of(portfolio, Trade, raw=True)

`create_view`

Create a SQL VIEW that extracts model fields from the JSON meta column into queryable columns.

coll.create_view(name, *models)

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	View name suffix. Final view name is `{collection}_view_{name}`
`*models`	`type[CentauroModel]`	—	One or more model classes whose fields define the view columns
`include_fields`	`list[str]`	`None`	Restrict the view to only these field columns
`overwrite`	`bool`	`False`	Drop and recreate the view if it already exists

Raises

ViewAlreadyExistsError — if the view exists and overwrite=False
ValueError — if include_fields contains unknown field names

# Basic view — all fields
coll.create_view("dashboard", Sensor)

# Restrict to a subset of columns (faster on wide models)
coll.create_view("lite", Sensor, include_fields=["unit", "location"])

# Multi-model union view
coll.create_view("all", Sensor, Alert)

# Replace an existing view
coll.create_view("dashboard", SensorV2, overwrite=True)

After creation, query with plain SQL — sql_select returns a Polars DataFrame:

df = coll.sql_select("""
    SELECT location, threshold
    FROM monitoring_view_dashboard
    WHERE unit = 'Celsius' AND threshold > 50
""")

`refresh_view`

Drop and recreate an existing view.

When include_fields is omitted (None), refresh_view introspects the existing view and reuses its current column configuration. Renamed fields are translated automatically — a column created under the old name maps to the new name via renamed_from() without manual intervention.

coll.refresh_view(name, *models)

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	View name suffix used when the view was created
`*models`	`type[CentauroModel]`	—	Updated model classes
`include_fields`	`list[str]`	`None`	Field subset for the refreshed view. When `None`, reuses the existing view's columns.

# Simplest form — reuses existing columns, translates renames
coll.refresh_view("dashboard", SensorV2)

# Explicitly change the column set
coll.refresh_view(
    "dashboard",
    SensorV2,
    include_fields=["unit", "location"],
)

`describe_views`

List all views belonging to this collection and their columns.

coll.describe_views()

Returns — dict[str, list[str]] mapping full view names to lists of model-field column names (excluding fixed columns id, name, write_time, edit_time).

coll.create_view("dashboard", Sensor)
coll.create_view("lite", Sensor, include_fields=["unit", "location"])

print(coll.describe_views())
# {
#   'monitoring_view_dashboard': ['unit', 'location', 'threshold'],
#   'monitoring_view_lite': ['unit', 'location'],
# }

`create_index`

Create an index on a field. Idempotent — safe to call on every application startup.

Automatically selects the correct index type:

Plain column index for built-in fields: id, name, write_time, edit_time
JSON expression index for any other name (field stored in the meta blob)

coll.create_index(field)

Parameters

Parameter	Type	Description
`field`	`str`	Field name to index

coll.create_index("unit")        # CREATE INDEX ... ON monitoring_objects (json_extract(meta, '$.unit'))
coll.create_index("write_time")  # CREATE INDEX ... ON monitoring_objects (write_time)
coll.create_index("unit")        # no-op — index already exists

Indexes live on the objects table, not on views — they are fully independent. See Indexing fields for performance.

`drop_index`

Drop an index by field name. Silent if the index does not exist.

coll.drop_index(field)

Parameters

Parameter	Type	Description
`field`	`str`	The field name used when the index was created

Raises — ValueError if field refers to the auto-managed name index, which CentauroDB requires for correct operation.

coll.drop_index("unit")          # drops idx_monitoring_unit
coll.drop_index("old_field")     # no-op if the index doesn't exist
coll.drop_index("name")          # raises ValueError — managed index

`list_indexes`

List all user-managed indexes on this collection's objects table. The internal name index is always excluded.

coll.list_indexes()

Parameters

Parameter	Type	Default	Description
`raw`	`bool`	`False`	Return full index names instead of field-name suffixes

Returns — list[str] sorted alphabetically.

coll.create_index("unit")
coll.create_index("write_time")

coll.list_indexes()
# ['unit', 'write_time']

coll.list_indexes(raw=True)
# ['idx_monitoring_unit', 'idx_monitoring_write_time']

All user-created indexes share the same naming convention and appear here.

`sql_select`

Execute a read-only SELECT (or WITH … SELECT) query and return the result as a Polars DataFrame. A convenience wrapper around engine.sql_select — no need to reference the engine directly.

df = coll.sql_select(sql, params)

Parameters

Parameter	Type	Default	Description
`sql`	`str`	—	A `SELECT` or `WITH … SELECT` query
`params`	`tuple`	`()`	Bind parameters for the query

Returns — polars.DataFrame

Raises — ValueError if the statement is not a SELECT/WITH query. ImportError if polars is not installed.

coll.create_view("dashboard", Sensor)

df = coll.sql_select("""
    SELECT location, AVG(threshold) AS avg_threshold
    FROM monitoring_view_dashboard
    WHERE unit = 'Celsius'
    GROUP BY location
""")
print(df)
# ┌──────────┬───────────────┐
# │ location ┆ avg_threshold │
# │ ---      ┆ ---           │
# │ str      ┆ f64           │
# ╞══════════╪═══════════════╡
# │ pelion   ┆ 80.0          │
# └──────────┴───────────────┘

Requires the polars extra: pip install centaurodb[polars]. See also engine.sql_select and engine.sql_execute for lower-level access and mutation queries.

TimeSeriesCollection

A dedicated collection type for CentauroModelSeries models. Creates both a {name}_objects table and a {name}_values table for storing (time, value) pairs per object. Shares most methods with Collection (write, read, query, update, refresh, delete, views, indexes, relations) with the following additions and overrides — count_objects and aggregate are inherited unchanged and fully available.

Constructor

from centaurodb import TimeSeriesCollection

coll = TimeSeriesCollection(engine, "stations")

Creates both stations_objects and stations_values tables.

`write_object` (override)

Inserts the object row and all obj.values in a single operation.

coll.write_object(obj)

Raises — TypeError if obj is not a CentauroModelSeries instance.

station = WeatherStation(
    city="Sparta",
    values=[
        CentauroValues(time="2025-06-01T08:00:00", value=22.5),
        CentauroValues(time="2025-06-01T09:00:00", value=24.1),
    ],
)
coll.write_object(station)
print(station.row.id)   # 1

`write_objects` (override)

Insert multiple time-series objects in a single atomic transaction. All objects must be CentauroModelSeries instances.

coll.write_objects(objs)

Raises — TypeError if any element is not a CentauroModelSeries.

metrics = [Metric(unit="kg"), Metric(unit="m")]
coll.write_objects(metrics)

`read_object_by_id` (override)

Read an object and, by default, hydrate its values list from the values table.

obj = coll.read_object_by_id(id, ModelClass)

Extra parameter

Parameter	Type	Default	Description
`hydrate_values`	`bool`	`True`	If `False`, skip loading values — `obj.values` will be empty

# Full read — object + all time-series values
station = coll.read_object_by_id(1, WeatherStation)
print(len(station.values))   # 2

# Metadata only — skip the values query
station = coll.read_object_by_id(1, WeatherStation, hydrate_values=False)
print(station.city)          # "Sparta"
print(station.values)        # []
station.df                   # raises ValuesNotHydratedError — see [Exceptions](/api/exceptions#valuesnothydratederror)

hydrate_values=False is useful when listing many objects for a summary view where you only need the metadata fields.

`read_objects` (override)

Read objects, optionally filtered by conditions, and hydrate time-series values.

results = coll.read_objects(WeatherStation)

Extra parameter

Parameter	Type	Default	Description
`hydrate_values`	`bool`	`True`	If `False`, skip loading values for each result

# All stations with values loaded
stations = coll.read_objects(WeatherStation)

# Metadata only — no values query per object
stations = coll.read_objects(WeatherStation, hydrate_values=False)

# Filtered with values
results = coll.read_objects(
    WeatherStation.fields.city == "Sparta",
    hydrate_values=True,
)

The time-series override does not accept limit / offset — only raw, hydrate_values, max_field, and min_field. To count without loading values, use the inherited count_objects().

`read_latest_object` (override)

Read the most recently edited object of a given type, with optional value hydration.

obj = coll.read_latest_object(ModelClass)

Extra parameter

Parameter	Type	Default	Description
`hydrate_values`	`bool`	`True`	If `False`, skip loading values

Returns — a deserialized model instance, a raw row when raw=True, or None if no objects of that type exist.

latest = coll.read_latest_object(WeatherStation)
if latest:
    print(latest.city)
    print(latest.df)

# Metadata only
latest = coll.read_latest_object(WeatherStation, hydrate_values=False)

`update_object` (override)

Update the object metadata and handle the values according to values_mode.

coll.update_object(obj, values_mode="append")

Extra parameter

Parameter	Type	Default	Description
`values_mode`	`str`	`"append"`	How to handle `obj.values` — see below

Values modes

append (default)

INSERT OR IGNORE — inserts new timestamps, skips duplicates. Use for incremental ingestion.

upsert

INSERT OR REPLACE / ON CONFLICT DO UPDATE — inserts new timestamps, overwrites existing ones. Use when values can be corrected.

replace

DELETE all existing values then bulk insert. Use for full replacement of a series.

station = coll.read_object_by_id(1, WeatherStation)
station.city = "Rome"
station.values = [CentauroValues(time="2025-06-01T10:00:00", value=26.3)]

# Add new timestamp without touching existing ones
coll.update_object(station, values_mode="append")

# Overwrite if the timestamp already exists
coll.update_object(station, values_mode="upsert")

# Delete all history and replace with the new list
coll.update_object(station, values_mode="replace")

`refresh` (override)

Re-read both the object metadata and its values from the database in place.

coll.refresh(obj)

station = coll.read_object_by_id(1, WeatherStation)

# ... another process appends new values ...

coll.refresh(station)
print(len(station.values))  # reflects the current DB state

Database tables

Collection type	Tables created
`Collection("monitoring")`	`monitoring_objects`
`TimeSeriesCollection("stations")`	`stations_objects`, `stations_values`

monitoring_objects

Column	Type	Description
`id`	`INTEGER`	Auto-incremented primary key
`name`	`TEXT`	Model's `__centauro_name__`
`write_time`	`DATETIME`	Timestamp of first insert
`edit_time`	`DATETIME`	Timestamp of last update
`meta`	`JSON` / `JSONB`	Full serialized model

stations_values

Column	Type	Description
`data_object_id`	`INTEGER`	Foreign key → `stations_objects.id` (CASCADE DELETE)
`time`	`DATETIME`	Timestamp of the data point (part of composite PK)
`value`	`REAL`	Numeric value
`write_time`	`DATETIME`	When this row was last written

Edit this pageorReport an issue

Engine

API reference for the Engine class.

Models

API reference for CentauroModel, CentauroModelSeries, CentauroValues, and RowFields.