Why ORMX?

pip install ormx-py

The PyPI distribution is ormx-py — the ormx and pyormx names there belong to unrelated packages. The import is ormx.

A full app in one file

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=150)


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. No mappers. No DTOs. No separate request/response models.

Feature status

4.5.0 — stable. 329 Python tests + 42 Rust tests passing.

Querying — the cookbook

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
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])

Many-to-many

# Simple — raw junction table, no extras.
class Article(Model):
    id: int = Field(primary_key=True)
    tags = ManyToMany("Tag", secondary="article_tags", through_local="article_id")

# Or with a junction model carrying extras (role, joined_at, …):
class Membership(Model):
    article_id: int = ForeignKey("articles.id")
    member_id: int = ForeignKey("members.id")
    role: str
    joined_at: int  # epoch seconds

class Article(Model):
    id: int = Field(primary_key=True)
    members = ManyToMany("Member", through="Membership")

# Bare targets (default):
articles = await Article.query().include("members").all()
# → articles[0].members: list[Member]

# With junction extras:
articles = await Article.query().include(("members", True)).all()
# → articles[0].members: list[(Member, Membership)]
#   so pair[1].role / .joined_at are addressable.

Composite indexes + composite foreign keys

from ormx import CompositeIndex, CompositeForeignKey

class OrderItem(Model):
    __tablename__ = "order_items"
    __indexes__ = [
        CompositeIndex("order_id", "created_at"),
        CompositeIndex("customer_id", "status", unique=True),
    ]
    __foreign_keys__ = [
        CompositeForeignKey(
            columns=("order_id", "product_id"),
            ref_table="catalog_entries",
            ref_columns=("order_id", "product_id"),
            on_delete="CASCADE",
        ),
    ]
    id: int = Field(primary_key=True)
    order_id: int
    product_id: int
    customer_id: int
    status: str
    created_at: int

The migration runner emits CREATE INDEX/ADD CONSTRAINT FOREIGN KEY (col1, col2) … automatically — no raw DDL needed.

Transactions — three flavours

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

# 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).

The migration primitives (discovery, DML rendering, state DDL) live in the ormx-ru Rust crate — shared across every ORMX SDK.

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. The router runs in Rust (md5-based deterministic hashing matching the Python strategy).

Architecture

┌─────────────────────────────────────┐
│  Python (Pydantic, asyncio)         │  ← User-facing API
├─────────────────────────────────────┤
│  PyO3 bridge (abi3)                 │  ← Zero-copy calls
├─────────────────────────────────────┤
│  ormx-core (Rust)                   │  ← Connection pool, bind/decode, router
│   ├─ Tokio runtime                  │
│   ├─ sqlx (Postgres / MySQL / SQLite)│
│   └─ PyO3 #[pyclass] bindings       │
└─────────────────────────────────────┘
            │
            ▼
┌─────────────────────────────────────┐
│  ormx-ru (Rust, published crate)    │  ← SQL discovery, DML, state DDL
│   ├─ discover    (file walking)     │
│   ├─ dml         (literal rendering)│
│   ├─ runner      (apply/rollback)   │
│   └─ check       (validation)       │
└─────────────────────────────────────┘

The Rust core lives in-tree at core/ and depends on the published ormx-ru crate for migration primitives — so anyone using ORMX from Rust, Go, JS, or Ruby shares the same migration engine.

Install

pip install ormx-py

Pre-built abi3 wheels for Linux (x86_64 + aarch64), macOS (Intel + Apple Silicon), and Windows. One pip install, no separate core package, no compiled extension to build locally.

For source builds (e.g. adding your own hooks), you need rustc 1.75+ — see RELEASING.md.

# Optional: install dev/test extras
pip install "ormx-py[dev]"
pip install "ormx-py[docs]"

Development

git clone https://github.com/shregar1/python.ormx.vexarr.com
cd python.ormx.vexarr.com

# Build the wheel locally (compiles ormx-core + ormx-ru from source)
maturin build --release
pip install --force-reinstall target/wheels/ormx_py-*.whl

# Run the suite
pytest                          # 317 Python tests
cd core && cargo test           # 42 Rust tests (10 ormx-core + 32 ormx-ru)

# Lint / type-check
ruff check .
mypy ormx/

Project layout:

python.ormx.vexarr.com/
├── ormx/                     # pure-Python package (Pydantic models, query builder)
├── ormx_core.pyi             # PyO3 type stubs (one file, ships in sdist)
├── core/                     # Rust extension (ormx-core crate)
│   ├── src/                  # lib.rs, logic.rs, migrate.rs, migrate_runner.rs
│   └── Cargo.toml            # depends on ormx-ru = "0.1" (crates.io)
├── examples/                 # rivex_basic.py, fastapi_basic.py, blog/
├── docs/                     # mkdocs material
└── pyproject.toml            # maturin build backend

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

Ecosystem

License

MIT.