Skip to content

no-positional-psycopg-row-escape

A positional Psycopg record must not escape its function unchanged.

Why

Returning a tuple-shaped database record couples callers to selected-column order. A named row factory keeps the database boundary explicit; scalar and locally transformed results remain valid.

Fix

For a returned record, use row_factory=class_row(Model) (prefer a validating model). Use scalar_row for scalar projections, or consume and transform a positional row inside the store.

Examples

Before — flagged A positional database record escapes unchanged
app/task_store.py
import psycopg
async def load(dsn: str):
async with await psycopg.AsyncConnection.connect(dsn) as conn:
async with conn.cursor() as cursor:
await cursor.execute("SELECT id, state FROM task")
return await cursor.fetchone()
After — preferred A returned record has a named row contract
app/task_store.py
from psycopg import AsyncConnection
from psycopg.rows import class_row
from pydantic import BaseModel
class TaskRow(BaseModel):
id: str
state: str
async def load(conn: AsyncConnection[tuple[object, ...]]):
async with conn.cursor(row_factory=class_row(TaskRow)) as cursor:
await cursor.execute("SELECT id, state FROM task")
return await cursor.fetchone()

Formerly: require-validated-row-factory