ORMX (Python)

Pydantic-native async ORM for Postgres, with a Rust core for the hot path and a real migration engine.

The headline trick: one class plays three roles — DB persistence object, framework request body, OpenAPI response model — without a translation layer. Drop into Rivex, FastAPI, Litestar, anything that knows Pydantic.

from rivex import Depends, Rivex
from ormx import Field, Model, TransactionMiddleware, connect, disconnect, get_db


class User(Model):
    __tablename__ = "users"
    id: int = Field(primary_key=True)
    email: str = Field(unique=True, db_index=True, pattern=r"^[^@\s]+@[^@\s]+$")
    name: str = Field(min_length=1, max_length=100)
    age: int = Field(ge=0, le=120)


app = Rivex()
app.add_middleware(TransactionMiddleware)


@app.on_event("startup")
async def init(): await connect("postgres://localhost/myapp")

@app.on_event("shutdown")
async def close(): await disconnect()


@app.get("/users/{id}", response_model=User)
async def get_user(id: int, db=Depends(get_db)):
    return await User.get(id)


@app.post("/users", response_model=User)
async def create_user(user: User, db=Depends(get_db)):
    return await user.save()

That's the entire stack.

Status

3.1.0 — stable. 171 tests passing.

Install

pip install ormx

ormx is pure-Python; the Rust core ships as a separate ormx-core package and installs transitively. Wheels are published for Linux (x86_64 + aarch64), macOS (Intel + Apple Silicon), and Windows. Source builds need rustc 1.75+.

Querying — the cookbook

ORMX query builders are chainable. Every chain method returns self; terminal methods (all, first, count, exists, scalar, values, update, delete, stream) execute the query and return.

# Basic WHERE
users = await User.query().where(User.age > 18).all()

# Chained — AND
adults = await (
    User.query()
    .where(User.age > 18)
    .where(User.email.like("%@example.com"))
    .order_by("name")
    .limit(50)
    .all()
)

# Count / exists / scalar
total = await User.query().count()
has_admin = await User.query().where(User.email == "admin@x.com").exists()

# Aggregates with GROUP BY + HAVING (HAVING is a raw SQL fragment)
buckets = await (
    User.query()
    .select("age")
    .group_by("age")
    .having("count(*) > 5")
    .all(raw=True)
)
# → [{"age": 30}, {"age": 31}, ...]

# DISTINCT and DISTINCT ON
unique_ages = await User.query().select("age").distinct().values("age")
latest_per_user = await (
    Event.query()
    .order_by("user_id").order_by("created_at", "desc")
    .distinct("user_id")
    .all()
)

# Bulk update / delete (skip per-row hooks)
n = await User.query().where(User.age < 13).update(age=13)
n = await User.query().where(User.deleted_at.is_not_null()).delete()

# Streaming — memory-bounded, server-side cursor
async for user in User.query().where(User.age > 50).stream(chunk_size=1000):
    process(user)

# Refresh a stale instance after an external write
await user.refresh()  # re-reads by PK

Relationships and eager loading

class User(Model):
    __tablename__ = "users"
    id: int = Field(primary_key=True)
    posts = Relationship("Post", back_populates="user")

class Post(Model):
    __tablename__ = "posts"
    id: int = Field(primary_key=True)
    user_id: int = ForeignKey("users.id", on_delete="CASCADE")
    title: str = Field(db_index=True)
    user = Relationship("User", back_populates="posts")


# One IN-query per included relation — no N+1.
users = await User.query().include("posts").all()
for u in users:
    print(u.name, [p.title for p in u.posts])

Transactions

Three flavours, pick what fits:

# 1. Explicit context manager
async with ormx.transaction():
    await User.create(name="alice")
    await Order.create(user_id=1, total=10)
# Rollback automatic on exception; commit on clean exit.

# 2. Unit-of-Work (batched flush at the end)
async with ormx.uow_session():
    user = User(id=1, name="alice", age=30)
    user.age = 31         # implicitly dirty
    await user.save()      # flushed at context exit, all in one tx

# 3. Per-request middleware (the most common shape)
app.add_middleware(TransactionMiddleware)
# Every handler wrapped automatically.
# Default: rollback on 5xx responses + raised exceptions; configurable.

Migrations

# Diff models vs DB, write a new timestamped migration file
ormx makemigrations

# Apply unapplied migrations in order
ormx migrate

# Roll back the most recently applied one
ormx migrate --rollback

# Drop orphan tables (off by default — safety against accidental data loss)
ormx makemigrations --allow-drop-tables

State lives in an ormx_migrations table the runner creates on first use. Each migration runs in its own transaction; partial-failure rolls back cleanly.

Detects: CREATE/DROP TABLE, ADD/DROP COLUMN, ALTER COLUMN TYPE (with USING), CREATE/DROP INDEX, ADD/DROP FOREIGN KEY, SET/DROP NOT NULL, CREATE SCHEMA for non-public-schema models. Multi-statement DDL + PL/pgSQL with dollar-quoted bodies work via the Postgres simple-query protocol — executor.execute_many(sql).

Multi-schema

class AuditEvent(Model):
    __tablename__ = "events"
    __schema__ = "audit"   # ← non-public schema; CREATE SCHEMA emitted automatically
    id: int = Field(primary_key=True)
    event_type: str

# Cross-schema FK
class AuditRef(Model):
    __tablename__ = "refs"
    event_id: int = ForeignKey("audit.events.id")

Sharding

import ormx_core
from ormx import set_router
from ormx.sharding import Router, HashSharding

# Connect each shard
await ormx.connect("postgres://shard0/...", shard_name="shard0", is_default=True)
await ormx.connect("postgres://shard1/...", shard_name="shard1")

# Install a routing strategy
set_router(Router(shards=["shard0", "shard1"], strategy=HashSharding()))

# Pass ``sharding_value`` and ORMX routes to the right shard
await User.query().where(User.id == 42).first(sharding_value=42)

Hash / range / geo / list / custom routing all supported via ormx_core.ShardingConfig.

Settings management

ORMX itself doesn't ship one — use pydantic-settings (standard) or write a small helper. Combined with Rivex's Settings base class:

from rivex import Settings

class Config(Settings):
    database_url: str
    debug: bool = False

cfg = Config.from_env()
await ormx.connect(cfg.database_url)

Upgrading from 2.x

See MIGRATION-3.0.md. TL;DR: type annotations on fields are now required (Pydantic uses them); ormx.fastapi was replaced by framework-agnostic primitives (TransactionMiddleware + get_db work with FastAPI, Rivex, Litestar — anything that knows async-generator dependencies).

Examples

• examples/rivex_basic.py — full CRUD with Rivex (one file)
• examples/fastapi_basic.py — same model, FastAPI
• examples/blog/ — full app: User + Post + Comment with relationships, eager loading, transactions, curl walkthrough

Documentation

Full docs at docs/ — mkdocs serve to read locally, auto-publishes to https://ormx.dev/docs via the docs workflow.

License

MIT.