Skip to content
Sarj Code Standards

no-psycopg-execution-in-loop

Use bounded bulk queries instead of executing Psycopg operations inside item loops.

SARJ560 Rule source Tests Catalog JSON

Why

Executing a database operation for each input item makes round trips scale with collection size, including loops that prepare each row before executing.

Fix

Collect inputs and use a bounded set-based query, typed bulk store operation or executemany batch. Preserve tenant checks, revision predicates, transaction boundaries, returned rows, ordering and deliberate sequential effects.

Examples

Before — flagged A connection executes once per item
app/store.py
from psycopg import AsyncConnection




async def remove(connection: AsyncConnection, rows):
    for row in rows:
        await connection.execute("DELETE FROM item WHERE id = %s", (row.id,))
After — preferred One query consumes the collected identifiers
app/store.py
from psycopg import AsyncConnection




async def remove(connection: AsyncConnection, rows):
    identifiers = [row.id for row in rows]
    await connection.execute("DELETE FROM item WHERE id = ANY(%s)", (identifiers,))