Use when the task is to migrate legacy Flask-SQLAlchemy .query. patterns to modern db.session.execute(sa.select(...)) style, or to standardize ORM query patterns. Do not use for schema changes, new queries, or business logic modifications.
Model.query.* patterns to db.session.execute(sa.select(...)) or db.session.get().scalar_one_or_none(), .scalars().all(), etc.)import sqlalchemy as sa where missing.query. calls should be modernized in the same passsandbox_pms_mvp/pms/extensions.py (confirms db import path)sandbox_pms_mvp/pms/models.py first 50 lines (confirms model import patterns)docs/audit-execution-plan.mdIncrementally migrate all legacy Model.query.* calls to the modern db.session.execute(sa.select(...)) pattern, one file at a time, preserving exact query semantics and verifying with the test suite.
app.py: 37seeds.py: 27admin_service.py: 23housekeeping_service.py: 23auth_service.py: 16front_desk_board_service.py: 14cashier_service.py: 13ical_service.py: 13staff_reservations_service.py: 11public_booking_service.py: 10availability_service.py: 10reporting_service.py: 9messaging_service.py: 9front_desk_service.py: 9provider_portal_service.py: 8room_readiness_service.py: 7communication_service.py: 7routes/front_desk.py: 7payment_integration_service.py: 6routes/staff_reservations.py: 6reservation_service.py: 4extras_service.py: 3routes/reports.py: 3Simple get-by-id:
# BEFORE
obj = Model.query.get(id)
# AFTER
obj = db.session.get(Model, id)
Filter + first:
# BEFORE
obj = Model.query.filter(Model.col == val).first()
obj = Model.query.filter_by(col=val).first()
# AFTER
obj = db.session.execute(sa.select(Model).where(Model.col == val)).scalar_one_or_none()
Filter + all:
# BEFORE
rows = Model.query.filter(cond).all()
rows = Model.query.filter_by(k=v).all()
# AFTER
rows = db.session.execute(sa.select(Model).where(cond)).scalars().all()
Order + all:
# BEFORE
rows = Model.query.order_by(Model.col.asc()).all()
# AFTER
rows = db.session.execute(sa.select(Model).order_by(Model.col.asc())).scalars().all()
Count:
# BEFORE
n = Model.query.filter(cond).count()
# AFTER
n = db.session.execute(sa.select(sa.func.count()).select_from(Model).where(cond)).scalar()
Join + filter:
# BEFORE
rows = Model.query.join(Other).filter(Other.col == val).all()
# AFTER
rows = db.session.execute(sa.select(Model).join(Other).where(Other.col == val)).scalars().all()
Exists check (first() used as truthy):
# BEFORE
if Model.query.filter(cond).first():
# AFTER
if db.session.execute(sa.select(Model).where(cond)).scalar_one_or_none():
Paginate (if used):
# BEFORE
page = Model.query.filter(cond).paginate(page=p, per_page=n)
# AFTER
stmt = sa.select(Model).where(cond)
total = db.session.execute(sa.select(sa.func.count()).select_from(stmt.subquery())).scalar()
rows = db.session.execute(stmt.offset((p-1)*n).limit(n)).scalars().all()
import sqlalchemy as sa must be present at the top of the file (check if already imported)from ..extensions import db or from .extensions import db (already present in most files)Pick one file. Start with the highest-count production file that has related test coverage. Priority order: service files first (business-critical), then route files, then seeds.py, then app.py.
Read the full file. Identify every .query. occurrence. Note the exact pattern for each (get, filter+first, filter+all, count, join, order_by, etc.).
Check imports. Verify import sqlalchemy as sa exists. Add it if missing (after other stdlib imports, before Flask imports).
Transform each occurrence using the patterns above. For each transformation:
first() is used in a truthy check, use scalar_one_or_none() (returns None or the object)all() returns a list, use .scalars().all() (returns a list)get() is used, prefer db.session.get(Model, id) (simplest)Do NOT change logic. The query result type and value must remain identical. If filter_by(status="active").all() returned a list, the replacement must also return a list.
Run targeted tests. After migrating a file, run the related test module:
python -m pytest sandbox_pms_mvp/tests/test_<related>.py -p no:cacheprovider -q
If no obvious test file exists, run the full suite.
Report the migration count. State how many .query. calls were migrated, how many remain in the file, and test results.
import sqlalchemy as sa added (yes/no).query. count across the codebase.query. from test files unless specifically asked (test patterns are lower priority).query. pattern does not map cleanly to the transformation patterns above, flag it and skipscalar_one_or_none() is used where first() was used (not scalar_one() which raises on None).scalars().all() is used where .all() was used (not .scalars() alone which returns an iterator).query. calls in the target file are replaced with modern db.session.execute() or db.session.get() patternsimport sqlalchemy as sa is present in the file