rowguard 0.5.0

Validation-first SQLAlchemy queries with Pydantic row validation.

RowGuard

RowGuard turns SQLAlchemy query results into validated Pydantic models.

By default, SQLRules (a required dependency) pushes supported model constraints into SQL so invalid candidates are filtered before fetch. Every row that is fetched is either an accepted model or an explicit rejection—never ignored after fetch.

Need to inspect invalid rows in Python? Pass use_sqlrules=False and on_reject="collect".

Use it when you already have SQLAlchemy Core tables/selects or ORM / SQLModel mapped classes and need typed reads with deterministic rejection handling. It is not an ORM and does not replace SQLAlchemy, Pydantic, or SQLModel—it validates reads over those stacks.

Status

Current release: 0.5.0 (Core + async + ORM/SQLModel). See Supported vs planned for what is shipped versus deferred.

Install

pip install rowguard                 # Core (includes SQLRules)
pip install "rowguard[async]"        # aiosqlite + greenlet for async
pip install "rowguard[sqlmodel]"     # SQLModel table-source support
pip install "rowguard[postgresql]"   # psycopg driver helper

Extra	When you need it
(none)	Sync Core / ORM reads
async	aselect / astream examples and async tests
sqlmodel	SQLModel mapped classes as table= / source=
postgresql	Optional PostgreSQL driver
dev	Contributors: pytest, ruff, mypy, …
docs	Sphinx documentation build

Requires Python 3.10+, Pydantic v2, SQLAlchemy 2.x, and SQLRules ≥1.0 (3.10–3.12 tested in CI; 3.13 untested). See the installation guide.

Quickstart

Full walkthrough: Quickstart.

1. Default path (library defaults)

Invalid candidates are filtered in SQL. rejected is empty.

from typing import Annotated

from pydantic import BaseModel, Field
from sqlalchemy import Column, Integer, MetaData, String, Table, create_engine
from sqlalchemy.orm import Session

import rowguard


class UserRead(BaseModel):
    id: int
    name: str
    age: Annotated[int, Field(ge=18)]


metadata = MetaData()
users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("age", Integer),
)

engine = create_engine("sqlite+pysqlite:///:memory:")
metadata.create_all(engine)

with engine.begin() as connection:
    connection.execute(
        users.insert(),
        [
            {"id": 1, "name": "Ada", "age": 37},
            {"id": 2, "name": "Legacy", "age": 12},
        ],
    )

with Session(engine) as session:
    result = rowguard.select(session=session, table=users, model=UserRead)
    print(result.models)    # (UserRead(id=1, name='Ada', age=37),)
    print(result.rejected)  # ()

2. Inspect rejections in Python

with Session(engine) as session:
    result = rowguard.select(
        session=session,
        table=users,
        model=UserRead,
        on_reject="collect",      # default is "raise"
        use_sqlrules=False,       # default is True
    )
    print(result.models)    # Ada
    print(result.rejected)  # Legacy failed age >= 18

See SQLRules pushdown and the FAQ.

Public API (0.5.0)

Function	Purpose
select / execute	Buffered validated reads
stream	Stream accepted models (no accepted-row buffer)
aselect / aexecute / astream	Async counterparts
validate_rows	Validate mappings without SQL
compile_plan	Inspect an ExecutionPlan without executing

Rejection policies: raise (default), collect, skip.

table= vs source=: use table= on select/stream (Core Table or mapped class). On execute with a projected Select, pass the mapped class as source=. Full parameter contracts: API guide · Python autodoc · Errors.

ORM / SQLModel guide · Streaming · Async · Performance · Upgrading.

Documentation

Start here → Installation → Quickstart.

• Docs home
• Supported vs planned
• Examples
• Changelog

Development

See Contributing and Security.

make install      # .[dev,async,sqlmodel] — matches CI
make all
python examples/sqlrules_default.py
python examples/basic.py

License

MIT