rowguard 0.4.0

Validation-first SQLAlchemy queries with Pydantic row validation.

RowGuard

RowGuard makes every SQLAlchemy query return validated Pydantic models—or explicit rejected rows. It does not silently drop bad data.

Use it when you already have SQLAlchemy Core tables/selects and need typed reads with deterministic rejection handling. It is not an ORM and does not replace SQLAlchemy or Pydantic.

Status

Current release: 0.4.0 (sync + async Core, streaming). See Supported vs planned for what is shipped versus deferred (ORM 0.5, callback/quarantine 0.6).

Install

pip install rowguard

Requires Python 3.10+, Pydantic v2, SQLAlchemy 2.x, and SQLRules ≥0.4. See the installation guide.

Quickstart

Full walkthrough: Quickstart.

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:
    # Disable SQLRules pushdown so invalid rows reach Pydantic and appear in rejected.
    result = rowguard.select(
        session=session,
        table=users,
        model=UserRead,
        on_reject="collect",
        use_sqlrules=False,
    )
    print(result.models)
    print(result.rejected)

    with rowguard.stream(
        session=session,
        table=users,
        model=UserRead,
        on_reject="skip",
        use_sqlrules=False,
    ) as stream:
        for model in stream:
            print(model)

With use_sqlrules=True (the default), supported constraints such as age >= 18 are pushed into SQL, so invalid candidate rows may never be returned. See SQLRules pushdown and the FAQ.

Public API (0.4.0)

Full reference: API guide · Python autodoc · Error catalog.

Function	Purpose
select(...)	Build and execute a table query with validation
execute(...)	Validate rows from an existing Select
validate_rows(...)	Validate mappings without SQL
compile_plan(...)	Compile an ExecutionPlan without executing
stream(...)	Stream validated models without buffering accepted rows
aselect(...)	Async select for AsyncSession / AsyncConnection
aexecute(...)	Async execute for AsyncSession / AsyncConnection
astream(...)	Async stream (AsyncStreamResult) without buffering accepted rows

Rejection policies: raise (default), collect, skip — see rejection policies.

Optional planning knobs: compiled_rules= (precompiled SQLRules), strict= (Pydantic), field_map= / column_map= (validated at plan time).

Streaming knobs: yield_per=, observers= (StreamObserver / BaseStreamObserver). Observers remain sync callables in 0.4. See the streaming guide.

Async note: only DB I/O is awaited. Pydantic validation runs on the event loop; heavy models can block. Prefer async with rowguard.astream(...) for cleanup. Install async extras with pip install rowguard[async]. Details: async guide.

Architecture

Pydantic Model
      │
      ▼
SQLRules
      │
      ▼
SQLAlchemy Query
      │
      ▼
Database
      │
      ▼
Row Adapter
      │
      ▼
Pydantic Validation
      │
      ├── Accepted Model
      └── Rejected Row

More detail: architecture overview · design philosophy · specification.

Documentation

• Docs home
• Start here
• Quickstart
• Supported vs planned
• Examples
• API
• Changelog
• Roadmap

Build docs locally:

pip install -e ".[docs]"
make docs
# open docs/_build/html/index.html

Development

See Contributing, Security, and Releasing.

pip install -e ".[dev,async]"
make all          # ruff + mypy + pytest --cov
python examples/basic.py
python examples/streaming.py
python examples/async_basic.py

License

MIT