sqlalchemy-data-access

SQLAlchemy 2.0 Async Data Access

Context

• Applies to database access in services and helpers using AsyncSession (see your project’s database module and get_db / session dependency).
• SQL injection, parameterized access, least privilege DB users: mdc:rules/common/security.mdc (Input Validation — ORMs/parameterization). This rule is the SQLAlchemy 2.0 async expression of those requirements in this repo.
• Migrations vs runtime: schema changes via Alembic as separate deploy steps — mdc:rules/common/component-technical-standards.mdc (Database & Data Management).

Core style (SQLAlchemy 2.0)

• Use select() with await session.execute(...), then scalar_one(), scalar_one_or_none(), scalars().all(), or await session.scalar(select(...)) as appropriate.
• Use session.get(Model, pk) for primary-key lookups when you want identity-map behavior.
• Do not use legacy Session.query() on async sessions.
• Prefer ORM entities and Column expressions in where / order_by; build dynamic filters by composing sqlalchemy.and_ / or_ and Condition objects — not by concatenating SQL.

Parameter binding and dynamic SQL

• Never concatenate raw SQL with user-controlled strings — see mdc:rules/common/security.mdc. Use Core/ORM expressions so values are bound as parameters.
• For LIKE / ILIKE with untrusted fragments, bind literals safely, for example:
  • column.ilike(func.concat("%", literal(user_fragment), "%"))
    (pattern often lives in a small helper such as _author_filter on a service module).
• For dynamic ORDER BY, use an allowlist of column attributes (like SORTABLE_FIELDS + getattr(Model, field)), not raw strings from the client.

Reads and shape of queries

• Pagination: Prefer limit / offset (or keyset pagination for very large tables). Run a separate count() query when the API needs total hits, as in paged list endpoints — avoid loading the full table to count rows.
• Counts: select(func.count()).select_from(Model).where(...) (or func.count(Model.id)) rather than len((await session.scalars(select(Model))).all()).
• Columns: Use select(Model.col1, Model.col2) when you intentionally need a narrow projection; map rows explicitly.
• N+1: If you add relationships, load them in the same round trip with selectinload() / joinedload() on the outer select(), or use a single statement with joins — do not access unloaded collections in a loop without an eager-load strategy.
• Large result sets: Never load unbounded all() for user-facing endpoints without limit. For batch jobs that scan many rows, use chunked queries (keyset/limit loops) or SQLAlchemy streaming APIs supported by your installed version, and keep transactions short.