Agimus Python SDK

The Python client for the Agimus Platform: the objects of your ontology, read and written through one fluent builder over object sets, in sync and async flavours.

Installation

pip install agimus

Requirements: Python 3.10+. pandas and polars are optional (to_pandas(), to_polars()).

Quick start

from agimus import AgimusClient

client = AgimusClient()  # reads AGIMUS_API_KEY and AGIMUS_BASE_URL

customers = client.objects("Customer")

# Objects in region EU with a rejected approval
rejected = customers.filter(region="EU").has("approvals", state="rejected")
rejected.count()
rejected.sort("-createdAt").all()

# The approvals of those customers, every row, as a DataFrame
df = rejected.pivot("approvals").to_pandas()

# One object, a write
customers.get("C1")
customers.create({"id": "C9", "name": "Acme"})

The ontology

Every element has a stable identifier:

• Entities (Customer, Order): apiName, PascalCase.
• Properties (customerId, createdAt): apiName, camelCase.
• Links: relationships between entities, with a forward and a reverse apiName.
• Categories: flat tags grouping entities (finance, hr); categoryId, snake_case.

Discover them with client.list_entities(), client.get_entity_schema("Customer"), client.list_links() and client.list_categories().

Authentication and configuration

API keys are created in the Agimus dashboard under Settings > API Access (agm_...) and inherit the permissions of their service user.

from agimus import AgimusClient, RetryPolicy

client = AgimusClient(
    api_key="agm_...",                  # or AGIMUS_API_KEY
    base_url="https://api.agimus.ai",   # or AGIMUS_BASE_URL; this is the default
    timeout=60.0,                       # seconds per request
    retry=RetryPolicy(max_attempts=4, budget=60.0),  # None disables retries
)

Use with AgimusClient() as client: to close the connection pool.

Object sets

client.objects("Customer") is the set of every customer. Each step returns a new set, so a builder can be forked; a terminal sends it to the platform, which computes everything server-side.

Steps

filter(*clauses, **kwargs) keeps the objects matching every clause. Keyword arguments are field=value (equality) or field__op=value:

customers.filter(status="active", amount__gte=100, region__in=["EU", "US"])
customers.filter(name__starts_with="A", email__is_null=True)

Values follow the property's type: numbers for numeric fields, datetime, date, time, Decimal and bytes are encoded for you. Combine clauses with the F helpers when and is not enough:

from agimus import F

customers.filter(F.or_(F.where(region="EU"), F.where(amount__gte=1000)), status="active")
customers.filter(F.not_(F.where(state="rejected")))

has(link, *clauses, **kwargs) keeps the objects with a related object matching the clauses (any related object when none are given); has_none(link, ...) keeps those with none. These are link predicates, evaluated by the platform as one correlated query, and they nest:

customers.has("approvals", state="rejected")
customers.has_none("orders")
customers.has("orders", F.any("items", sku="X1"))

pivot(link) replaces the set by the related objects; the set's type becomes the link's target:

customers.filter(region="EU").pivot("approvals").filter(state="rejected").all()

with_count(link, ...), with_count_distinct(link, field, ...), with_sum, with_avg, with_min, with_max(link, field, ...) and with_value(link, field) compute one value per object over its related objects, optionally over a filtered subset (sum and avg take numeric fields, min and max orderable scalars: never an array, a boolean, JSON or bytes). The output is named <link>_<op> unless alias= says otherwise, and can be used in later filters, sorts, groups, metrics and projections:

customers.with_count("approvals", alias="rejections", state="rejected") \
         .with_sum("orders", "total", alias="revenue") \
         .filter(rejections__gte=2).sort("-revenue").fields("id", "revenue").all()

top(by, order, n=1) keeps the first n objects of every partition of by under order, so "each customer's latest approval" is a pivot and a top:

customers.pivot("approvals").top(by=["customerId"], order=["-approvedAt"]).filter(state="rejected")

Chains are bounded to three hops (pivots, link predicates, computed values and expands together).

Shape

sort("-createdAt", "name"), fields("id", "name"), expand("owner", "orders.items") (related objects embedded in each page row, up to three hops), page_size(n) (1 to 10,000, default 1,000), limit(n) (stop all() and iter() after n rows).

The plane

Every entity with a backing dataset is served from the warehouse; an entity with the operational tier is also served from Postgres (an entity created without a dataset lives on Postgres only). By default the platform chooses: Postgres for a chain every entity of which is operational and the request is light, the warehouse engine otherwise. plane() makes the choice yours, on every terminal, Arrow included:

customers.plane("operational").filter(id__in=ids).all()   # Postgres, milliseconds, indexes
customers.plane("analytical").has("orders").to_pandas()   # the engine at the applied snapshot, any size

A hint is a contract, never a silent fallback: operational on a chain with an entity that has no operational tier, or analytical on an entity whose warehouse twin is not servable yet, is a ValidationError (object_store.query.cross_plane_unavailable) answered before any query runs, naming the entity and the reason. The data is the same on both planes: edits are visible immediately on either, and the analytical read of an operational entity is pinned to the snapshot Postgres has applied. A page walk keeps the plane it started on.

Terminals

Cursors are opaque and signed; a page walk is pinned to the snapshot it started on, and refreshed says the platform re-pinned it after that snapshot became unreadable (on page(), on the iterator, on all()'s list).

Bulk: Arrow

to_arrow() delivers the objects of a set as one Arrow table over HTTPS: every row, api names, the platform's types, ordered only under sort(), inline below the platform's Arrow cap (64 MB; a QuotaError above it). It is the way to move data out of the platform; a page walk over millions of rows is not. There is no dataset surface: every read is an entity of the ontology.

table = customers.has("approvals", state="rejected").fields("id", "region").to_arrow()
df = customers.pivot("orders").sort("-createdAt").to_pandas()

expand is not available on Arrow: pivot to the related objects instead.

Aggregation

from agimus import F, G, M

approvals = client.objects("Approval")
result = approvals.aggregate(
    metrics=[M.sum("amount"), M.count(alias="n"), M.max("approvedAt", alias="last"),
             M.percentile("amount", 0.5, alias="p50")],
    group_by=["state", G.grain("approvedAt", "month"), G.ranges("amount", [20, 100])],
    having=F.having(n__gte=3),
    sort=["-sum_amount"],
    limit=100,
)
for row in result:
    print(row["state"], row["approvedAt_month"], row["sum_amount"])

Metrics: count (a plain count, or of a field's non-null values), count_distinct, sum, avg, min, max, first, last, percentile(field, quantile). The default alias is op_field (sum_amount) or op. Groups: a field as is, G.grain(field, granularity) for year, quarter, month, week, day or hour (the key is field_granularity, a UTC timestamp, midnight for a date field), G.ranges(field, edges) for numeric buckets (the key is field_range with labels <20, [20,100), >=100). first, last, sum, avg, min and max take scalar fields, never arrays. having is one clause on a metric alias or a field grouped as is (not a grain or range key), on eq, ne, gt, gte, lt or lte. Aggregations are exact on every engine and return at most 10,000 groups.

One object

customers.get("C1", fields=["id", "name"], expand=["owner"])
customers.get_or_none("C1")            # None only when the object does not exist
customers.batch_get(["C1", "C2"])      # up to 1,000 keys; the objects found, ordered by primary key
customers.related("C1", "approvals")   # a set: the approvals of C1, on the builder's plane and page size
customers.related("C1", "approvals").count()

Keys may be strings or integers; a key containing / is fine.

Writes

Writes need the operational tier; a write on an entity without it is refused typed (object_store.entity.not_operational).

customers.create({"id": "C9", "name": "Acme", "since": datetime.date(2026, 1, 2)})
customers.update("C9", {"name": "Acme Ltd"})
customers.upsert("C9", {"name": "Acme Ltd"})  # creates (every required field) or updates (editable fields only)
customers.delete("C9")
customers.batch([
    {"op": "create", "data": {"id": "C10", "name": "Beta"}},
    {"op": "update", "pk": "C9", "data": {"name": "Acme"}},
    {"op": "delete", "pk": "C8"},
])  # up to 100 operations; each succeeds or fails on its own, the result names every outcome

A write response is the row as the platform now serves it, typed like a read and limited to the properties your key can read.

Types

The SDK reads each entity's schema once (cached for five minutes) and decodes every value to its Python type, on pages, point reads, computed values, aggregates and expanded objects alike. The same types come back from to_arrow().

Two differences between the JSON page and the Arrow table: JSON-family values are objects on a page and canonical JSON text on Arrow; a decimal stored with more scale than its column declares is quantized on Arrow and served as stored on a page. Computed values follow the platform's rule: counts are int, avg is float, sum is the field's type widened (a decimal sum stays Decimal), min, max and value the field's own type. Point reads also carry the system columns (_created_at and the like) as text.

Errors and retries

Every failure is one AgimusError with the platform's envelope: code, message, category, status, request_id, details, and retry_after (seconds) when the platform said when to try again.

from agimus import (
    AgimusError, AuthenticationError, AccessDeniedError, NotFoundError, ValidationError,
    ConflictError, QuotaError, RateLimitError, ServerError, ConnectionFailed, RequestTimeout,
)

try:
    customers.get("C1")
except NotFoundError as e:
    print(e.code)          # object_store.object.not_found
except ConflictError as e:
    print(e.retry_after)   # the store is resizing or not ready yet; seconds to wait

The client retries on its own inside RetryPolicy(max_attempts=4, budget=60.0): a refusal carrying Retry-After (the rate limit, a busy executor or Arrow slot, a resizing or not-yet-ready store) is resent after exactly that wait when it fits the budget and raised at once otherwise, since such a refusal is answered before anything is applied; a connection failure is retried with backoff for every method (nothing was sent); a timeout only for reads, because a write may have applied. retry=None disables all of it.

Async

from agimus import AsyncAgimusClient

async with AsyncAgimusClient() as client:
    rejected = client.objects("Customer").has("approvals", state="rejected")
    print(await rejected.count())
    async for customer in rejected.sort("id"):
        ...
    table = await rejected.to_arrow()

The same builder, every terminal awaited.

Schema discovery

client.list_entities(categories=["finance"])
client.get_entity_schema("Customer")     # properties (types, declarations, PK), links, categories
client.get_properties("Customer")
client.get_property("Customer", "name")
client.get_primary_key("Customer")
client.get_links("Customer")
client.list_links()
client.get_link_schema("approvals")
client.list_categories()
client.get_category("finance")
client.me()                              # the key's tenant, scope and rate limit

License

MIT