CRUD with select(), Sessions and Transactions

SQLAlchemy 2.0 uses select() statements executed with session.execute()/session.scalars(). Add objects with session.add(), flush() to get generated IDs, and commit once per unit of work. Use session.get() for primary-key lookups, and bulk update()/delete() statements for set-based changes. Keep queries in repository/service functions.

from sqlalchemy import select, update


async def get_user(db: AsyncSession, user_id: int) -> User | None:
    return await db.get(User, user_id)


async def list_orders(db: AsyncSession, user_id: int, limit: int, after_id: int | None) -> list[Order]:
    stmt = select(Order).where(Order.user_id == user_id).order_by(Order.id.desc()).limit(limit)
    if after_id is not None:
        stmt = stmt.where(Order.id < after_id)
    return list(await db.scalars(stmt))


async def create_order(db: AsyncSession, user_id: int, total_cents: int) -> Order:
    order = Order(user_id=user_id, total_cents=total_cents)
    db.add(order)
    await db.flush()          # assigns order.id; commit happens in the session dependency
    return order


async def deactivate_stale(db: AsyncSession) -> int:
    result = await db.execute(update(User).where(User.last_login < cutoff).values(active=False))
    return result.rowcount

Common pitfalls

• Committing inside every helper, breaking atomicity of multi-step operations.
• Loading all rows then filtering in Python.
• Using legacy session.query() style in new code.

Key points

• select() + scalars()
• flush for IDs, commit once per unit of work
• Set-based updates for bulk changes

Further reading: SQLAlchemy: ORM querying guide