Async¶

RowGuard ships first-class async APIs over SQLAlchemy AsyncSession and AsyncConnection.

Sync	Async
`select`	`aselect`
`execute`	`aexecute`
`stream`	`astream` → `AsyncStreamResult`

pip install "rowguard[async]"

Pass exactly one of session= or connection= (AsyncSession / AsyncConnection).

Default path¶

result = await rowguard.aselect(
    session=session,
    table=users,
    model=UserRead,
)
# use_sqlrules=True, on_reject="raise" by default

Inspect rejections¶

result = await rowguard.aselect(
    session=session,
    table=users,
    model=UserRead,
    on_reject="collect",
    use_sqlrules=False,
)
print(result.rejected)

Streaming¶

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

astream() returns immediately; work starts on async with / async for.

Event-loop note¶

Only database I/O is awaited. Pydantic validation runs synchronously on the event loop and can block under heavy models. See Performance.

Not in 0.6¶

Thread-pool wrappers around sync APIs
Required asyncpg CI matrix (sqlite+aiosqlite is the async CI driver)
Plugin-registered quarantine providers (0.7)