agnes (Python)

Python port of agnes-library — a type-hinted database toolkit with a Rust core (pool, SQL parser, self-invalidating cache) exposed via PyO3. Same schema DSL, query builder, relations and cache as the TypeScript version; synchronous API.

Install

pip install agnes-library      # imports as `agnes`

Build from source (needs Rust + maturin):

cd agnes-py && maturin develop --release

Schema

table(def, "physical_name"). Columns: int_, bigint, text, bool_, float_, bytes_, json_ (trailing _ avoids shadowing builtins). Modifiers: .primary(), .nullable(), .default(v), .autoincrement(), .index("n"), .unique_index("n"). Relations: one(target, local_key, target_key, OnAction, OnAction), many(target, fk).

from agnes import table, int_, text, bool_, one, many, OnAction

schema = {
    "user": table({
        "id": int_("id").primary().autoincrement(),
        "name": text("name").index("name_idx"),
        "email": text("email").unique_index("email_idx"),
        "active": bool_("active").default(True),
        "posts": many("post", "userId"),
    }, "users"),
    "post": table({
        "id": int_("id").primary().autoincrement(),
        "userId": int_("user_id"),
        "user": one("user", "userId", "id", OnAction.NONE, OnAction.CASCADE),
    }, "posts"),
}

Grouped (multi-schema) form flattens to a dotted key you select by — the relation target must use that dotted key:

schema = {"legislativo": {"etapas": table({...}, "legislativo.etapas")}}
# db.select("legislativo.etapas"); many("legislativo.etapas", "...")

Client

from agnes import AgnesClient, eq, gt, query

db = AgnesClient.create(
    {
        "driver": "postgres",            # "postgres" | "mysql" | "sqlite"
        "url": "postgres://user:pass@host/db",
        # connection pool tuning (all optional):
        "max_connections": 10,           # hard cap (default 10)
        "min_connections": 0,            # kept warm while idle (default 0)
        "acquire_timeout_secs": 30,      # wait for a free connection
        "idle_timeout_secs": 600,        # close after this idle time
        "max_lifetime_secs": 1800,       # recycle after this lifetime
        "strip_timezone": True,          # optional: timestamps as naive ISO (no offset)
        "cache": {"enabled": True, "wal_path": ".agnes/cache.wal"},
        # read/write splitting (master/slave), all optional:
        "replicas": ["postgres://replica-1/db", "postgres://replica-2/db"],
        "master_read_penalty": 100,      # bias reads toward replicas (default 100)
        "replica_cooldown_secs": 5,      # skip a replica this long after it errors
    },
    schema,
)
# With `replicas`, `url` is the write master: writes/transactions go there,
# reads are load-balanced to the least-busy node (master included, penalized).

U = db._schema["user"].c            # column accessor: U.age, U["age"]

rows = (db.select("user")
          .where(gt(U.age, 18), eq(U.active, True))
          .order_by(U.name, "asc")
          .limit(50)
          .ttl(60)                  # cache 60s, auto-invalidated on write
          .all())

with_posts = db.select("user").include({"posts": query().limit(3)}).all()

# pagination: limit/offset, or page(page, per_page) (1-based) — pair with order_by
page3 = db.select("user").order_by(U.id).page(3, 20).all()          # rows 41–60
same = db.select("user").order_by(U.id).limit(20).offset(40).all()

# choosing columns — two styles (`from_` has a trailing _ since `from` is a keyword):
db.select("user").all()                         # table-first, all columns
db.select().from_("user").all()                 # projection-first, all columns
db.select("name", "email").from_("user").all()  # only these columns
db.select().from_("user").omit("password").all()  # everything except password

# read-your-writes: force this read onto the master (skips lagging replicas)
fresh = db.select().from_("user").fresh_read().all()

# stream large results row-by-row (constant memory; server-side cursor on Postgres)
for user in db.select("user").where(gt(U.age, 18)).stream(batch_size=1000):
    process(user)
# not available in a transaction; incompatible with include()

db.insert_into("user").values({"name": "Ana", "age": 30})

# bulk insert — one statement, auto-chunked to the param limit; missing keys → NULL
db.insert_into("user").values([{"name": "Ana"}, {"name": "Bia", "age": 20}])

# upsert — on_conflict(*cols) with merge()/ignore()
db.insert_into("user").on_conflict(U.id).merge().values({"id": 1, "name": "Ana"})
db.insert_into("user").on_conflict(U.email).merge(U.name).values({"email": "a@x", "name": "Ana"})
db.insert_into("user").on_conflict(U.id).ignore().values({"id": 1, "name": "Ana"})

db.update("user", {"active": False}).where(eq(U.id, 1)).run()
db.delete_from("post").where(eq(db._schema["post"].c.id, 2)).run()

# returning() — get the affected rows back (Postgres/SQLite; raises on MySQL)
created = db.insert_into("user").returning().values({"name": "Ana"})
updated = db.update("user", {"age": 31}).where(eq(U.id, 5)).returning(U.id, U.age).run()
deleted = db.delete_from("post").where(eq(P.id, 9)).returning().run()

# raw SQL fallback
db.query("SELECT * FROM users WHERE age > $1", [18], {"ttl": 30})
db.mutate("UPDATE users SET active = $1 WHERE id = $2", [False, 1])

Schema push

Create the tables and indexes the schema describes — idempotent (CREATE ... IF NOT EXISTS), dependency-ordered so foreign-key targets come first. Only creates what's missing; never alters or drops (diff migrations are separate).

db.push_schema()          # create missing tables + indexes
sql = db.schema_ddl()     # inspect the statements (list[str])

Types map per-dialect; .autoincrement() → SERIAL / AUTO_INCREMENT / INTEGER PRIMARY KEY AUTOINCREMENT; one(...) relations emit FOREIGN KEY constraints with their on-update/delete actions.

Transactions

Interactive transactions, Prisma-style. db.transaction(fn) calls fn(tx) with the same query API on one connection; commits when it returns, rolls back and re-raises if it raises.

def transfer(tx):
    tx.update("account", {"balance": 60}).where(eq(acc.id, 1)).run()
    tx.update("account", {"balance": 40}).where(eq(acc.id, 2)).run()
    # raise here → both updates roll back

db.transaction(transfer)

Operators (leaves): eq, neq, gt, gte, lt, lte, like, ilike, in_array(col, [...]), not_in_array(col, [...]), is_null(col), is_not_null(col), between(col, lo, hi). Combine/nest with and_(...), or_(...), not_(...). where(a, b) means a AND b.

from agnes import eq, or_, in_array, is_null, not_

db.select("user").where(
    eq(U.active, True),
    or_(in_array(U.role, ["admin", "mod"]), is_null(U.role)),
    not_(eq(U.age, 0)),
).all()

Aggregations — count, sum_, avg, min_, max_ with .group_by(...) and .having(agg, op, value); terminal .aggregate({...}) returns one row per group:

from agnes import count, sum_, avg

db.select("order") \
  .where(gt(O.total, 0)) \
  .group_by(O.user_id) \
  .having(sum_(O.total), ">", 100) \
  .aggregate({"spent": sum_(O.total), "orders": count(), "avg_total": avg(O.total)})

count() with no argument is COUNT(*). (sum_/min_/max_ are underscored to avoid shadowing Python builtins.)

query() (for include) has .where, .order_by, .limit, .select(*cols), .type("left"|"inner"). SQL joins: .left_join/.inner_join/.right_join/.full_join("tbl", on(a, b)).

Prior art

The query API is inspired by Drizzle (fluent builder, eq/gt-style condition helpers) and Prisma (typed relations, interactive transactions). The feel is deliberately familiar, but the entire implementation is original work written from scratch — no source code from those projects is used or derived. "Drizzle" and "Prisma" are trademarks of their respective owners.

License

MIT OR Apache-2.0