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.6.0 (Core + async + ORM/SQLModel + rejection platform). 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

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

Rejection policies: raise (default), collect, skip, log, callback, quarantine — plus optional max_rejections / max_rejection_rate. See Rejection policies.

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