---
name: tap-sqlalchemy2-migration
description: SQLAlchemy 2.0 Migration Skill — Generic TAP OWS Service
---

# SQLAlchemy 2.0 Migration Skill — Generic TAP OWS Service

## Purpose

Use this skill whenever you need to migrate **any** `tap-ows-service` project
(e.g. `ows-payment`, `ows-payee`, `ows-*`) from SQLAlchemy 1.x to SQLAlchemy 2.0.

Covers:
- Identifying and replacing every SA 1.x pattern with its SA 2.0 equivalent
- Enforcing the strict DB-layer architecture rule (models/repository only — never logic)
- Updating tests
- Bumping dependencies safely

> **Notation used throughout this skill:**
> - `<pkg>` — the top-level Python package of the service (e.g. `payment`, `payee`)
> - `SomeModel` / `YourModel` — stand-ins for the actual ORM model class in your service
> - File paths are written as `<pkg>/models/`, `<pkg>/logic/`, `<pkg>/repository/` —
>   substitute the real package name when applying changes.

---

## 0. Architecture Rule: DB Code Belongs in Models / Repository — Never in Logic

> **This rule takes priority over all SA 2.0 pattern replacements below.**

The codebase is strictly layered:

| Layer | Allowed DB operations | Example |
|---|---|---|
| `<pkg>/models/` | `db.session.execute()`, `select()`, `update()`, `delete()`, `func.*`, `db.session.add_all()` | classmethods on the ORM model |
| `<pkg>/repository/` | Same as models — used for queries too complex to sit on the model class | standalone functions |
| `<pkg>/logic/` | **None of the above.** Call model classmethods or repository functions only. | `SomeModel.get_paginated(limit, offset)` |

**Logic modules must not import or use:**
- `db.session.execute()` / `db.session.add_all()` / `db.session.query()`
- `select()`, `update()`, `delete()`, `func`, `text`
- Any raw SQLAlchemy construct

**If you encounter `db.session.*` in a logic file during migration, do not just swap
the SA 1.x call for its SA 2.0 equivalent in-place. Instead, move the DB call into
the model or repository and expose a named classmethod / function that the logic
layer calls.**

### Anti-pattern (wrong — DB code in logic)
```python
# <pkg>/logic/some_resource.py  ❌
from sqlalchemy import func, select

def get_items(limit, offset):
    query = SomeModel.base_list_query()
    items = db.session.execute(query.offset(offset).limit(limit)).mappings().all()
    count = db.session.execute(
        select(func.count()).select_from(query.subquery())
    ).scalar_one()
    return Items(items, count)
```

### Correct pattern (DB code in model/repository, logic just calls it)
```python
# <pkg>/models/some_resource.py  ✅
@classmethod
def get_paginated(cls, limit: int, offset: int):
    """Return (items, total_count) for the resource list."""
    stmt = cls.base_list_query()
    items = db.session.execute(stmt.offset(offset).limit(limit)).mappings().all()
    count = db.session.execute(
        select(func.count()).select_from(stmt.subquery())
    ).scalar_one()
    return items, count

# <pkg>/logic/some_resource.py  ✅
def get_items(limit: int, offset: int) -> Items:
    items, count = SomeModel.get_paginated(limit, offset)
    return Items(items, count)
```

---

## 1. Dependency Changes (`pyproject.toml`)

Apply the following bumps. Exact patch versions may vary — always use the latest
non-breaking patch within the constraint shown.

```diff
-    "Flask-SQLAlchemy~=3.0.x",
+    "Flask-SQLAlchemy~=3.1.1",

-    "SQLAlchemy<2",
+    "SQLAlchemy~=2.0.50",

# abacus-common-logic must be ≥5.15.0 for SA 2.0 compatibility:
-    "abacus-common-logic~=5.14.x",
+    "abacus-common-logic~=5.15.0",
```

After editing `pyproject.toml`:
```bash
uv sync && uv lock
```

---

## 2. Blocking Changes (SA 2.0 will not start without these)

### 2.1 `db.engine.execute()` → `engine.connect()` + `text()`

Applies to every raw-SQL fixture in `tests/conftest.py` and
`tests/integration/conftest.py`.

**Add import:**
```python
from sqlalchemy import text
```

**Add helper at module level (both conftest files):**
```python
def _exec_sql(sql):
    """Execute raw SQL using SA2-compatible engine.connect()."""
    with db.engine.connect() as conn:
        conn.execute(text(sql))
        conn.commit()
```

**Replace every call:**
```python
# Before
db.engine.execute(sql)

# After
_exec_sql(sql)
```

### 2.2 `select([list])` → `select(*args)` (remove list brackets)

SA 2.0 removed the positional-list form of `select()`.

```python
# Before
select([col1, col2])
select(['*'])
select([literal(1)])
select([literal_column('alias.some_column_id')])

# After
select(col1, col2)
select(literal_column('1'))   # only for EXISTS/subquery checks
select(literal(1))
select(literal_column('alias.some_column_id'))
```

Special case — `select(['*'])` used inside `.exists()` / subquery existence checks:
```python
# Before
select(['*']).where(SomeModel.id == id_val)

# After — use a concrete column or literal_column('1')
select(literal_column('1')).where(SomeModel.id == id_val)
```

**Where to look:** Search every file under `<pkg>/models/` and `<pkg>/repository/`
for `select([`.

---

## 3. Deprecated Patterns (emit SA 2.0 warnings — must fix)

### 3.1 `Model.query` → `db.session.execute(select(Model)...)`

The `Model.query` shortcut is the legacy Query API. Replace all usages.

```python
# ── fetch all ──────────────────────────────────────────────────────────────
# Before
cls.query.filter(cls.deleted_at.is_(None)).all()
# After
db.session.execute(select(cls).where(cls.deleted_at.is_(None))).scalars().all()

# ── fetch first ────────────────────────────────────────────────────────────
# Before
cls.query.filter_by(code=code).first()
# After
db.session.execute(select(cls).where(cls.code == code)).scalars().first()

# ── count ──────────────────────────────────────────────────────────────────
# Before
cls.query.filter(...).count()
# After
db.session.execute(
    select(func.count()).select_from(select(cls).where(...).subquery())
).scalar_one()

# ── returning a statement for chaining ─────────────────────────────────────
# Before
cls.query.filter_by(name=name)   # returns a Query
# After — return a select() statement instead
select(cls).where(cls.name == name)

# ── chained offset/limit ───────────────────────────────────────────────────
# Before
query.order_by(cls.col).offset(n).limit(m).all()
# After
db.session.execute(
    select(cls).order_by(cls.col).offset(n).limit(m)
).scalars().all()

# ── filter() chained on a select() ────────────────────────────────────────
# Before (Query API)
query.filter(cls.col == val)
# After (Core select)
stmt.where(cls.col == val)
```

> **Pattern for classmethods that previously returned a `Query` object:**
> Change them to return a `select()` statement **and** add a companion execution
> classmethod so callers never need `db.session.execute()` directly:
>
> ```python
> # model — keep the statement builder
> @classmethod
> def filter_active(cls):
>     return select(cls).where(cls.deleted_at.is_(None))
>
> # model — add an executor so logic/repository don't touch db.session
> @classmethod
> def get_active(cls):
>     return db.session.execute(cls.filter_active()).scalars().all()
> ```
>
> Logic files call `SomeModel.get_active()`, never `db.session.execute(...)` directly.
> Repository files may use `db.session.execute()` directly for complex queries.

**Required imports (model/repository files only):**
```python
from sqlalchemy import func, select
```

### 3.2 `db.session.query()` → `db.session.execute(select(...)...)`

```python
# Before — multi-column aggregate query
db.session.query(
    cls.currency_code,
    func.count(cls.account_id).label('account_count'),
).filter(cls.deleted_at.is_(None)).group_by(cls.currency_code).all()

# After — use .mappings().all() to get RowMapping dicts (drop-in for old namedtuple rows)
db.session.execute(
    select(
        cls.currency_code,
        func.count(cls.account_id).label('account_count'),
    )
    .where(cls.deleted_at.is_(None))
    .group_by(cls.currency_code)
).mappings().all()

# Before — aggregate returning a single row
db.session.query(
    func.coalesce(func.sum(cls.amount), 0).label('amount'),
).filter(cls.event_id == event_id).one()

# After
db.session.execute(
    select(
        func.coalesce(func.sum(cls.amount), 0).label('amount'),
    ).where(cls.event_id == event_id)
).one()
```

> **Key difference:** When a query returned ORM model instances, use `.scalars().all()`.
> When it selected individual columns or aggregates, use `.mappings().all()` (returns
> `RowMapping` — dict-like) or `.one()` / `.scalar_one()`.

**Where to look:** Search every file under `<pkg>/models/` for `db.session.query(`.

### 3.3 `.query.filter(...).update()` → `db.session.execute(update(...)...)`

```python
# Before
cls.query.filter(
    cls.foreign_key_id == fk_id,
    cls.deleted_at.is_(None),
).update(
    {
        cls.deleted_at: cls.current_timestamp(),
        cls.deleted_by: get_flask_user_id(),
    },
    synchronize_session=False,
)

# After — synchronize_session moves to .execution_options()
#        values() uses keyword args (column name strings), NOT {cls.col: val} dicts
from sqlalchemy import update

db.session.execute(
    update(cls)
    .where(
        cls.foreign_key_id == fk_id,
        cls.deleted_at.is_(None),
    )
    .values(
        deleted_at=cls.current_timestamp(),
        deleted_by=get_flask_user_id(),
    )
    .execution_options(synchronize_session=False)
)
```

> **Note on `.values()` dict form:** Prefer keyword arguments (`deleted_at=val`) as
> the SA 2.0 idiomatic style. Mapped attribute dict keys (`{cls.deleted_at: val}`)
> still work in Core `update().values()` in SA 2.0 — only the legacy ORM
> `Query.update()` dict form was removed. Use kwargs to avoid ambiguity.

When the method returns a `rowcount`, capture `result.rowcount`:
```python
result = db.session.execute(update(cls).where(...).values(...).execution_options(...))
return result.rowcount
```

### 3.4 `.query.filter(...).delete()` → `db.session.execute(delete(...)...)`

```python
# Before
cls.query.filter(
    cls.foreign_key_id == fk_id
).delete(synchronize_session=False)

# After
from sqlalchemy import delete

db.session.execute(
    delete(cls)
    .where(cls.foreign_key_id == fk_id)
    .execution_options(synchronize_session=False)
)
```

### 3.5 `db.session.bulk_save_objects()` → `db.session.add_all()`

All worksheet / batch-create logic files commonly used `bulk_save_objects`. The
replacement is `add_all` — but **the call must live in the model or repository layer**,
not in logic.

```python
# <pkg>/models/some_resource.py  ✅
@classmethod
def bulk_create(cls, instances):
    db.session.add_all(instances)
    db.session.commit()

# <pkg>/logic/some_resource.py  ✅
def create_resources(data):
    instances = [SomeModel(**row) for row in data]
    SomeModel.bulk_create(instances)
```

> For high-volume inserts where no ORM events (`@validates`) exist, the Core INSERT
> approach can be used inside the model/repository:
> ```python
> from sqlalchemy.dialects.mysql import insert as mysql_insert
>
> # Strip SA-internal state key — obj.__dict__ includes '_sa_instance_state'
> db.session.execute(
>     mysql_insert(SomeModel),
>     [{k: v for k, v in obj.__dict__.items() if not k.startswith('_sa_')} for obj in instances],
> )
> ```
> Alternatively, expose a `to_dict()` method on the model and use that instead of `__dict__`.

**Where to look:** Search `<pkg>/logic/` for `bulk_save_objects`. Every occurrence
must be moved into a model classmethod or repository function.

### 3.6 `db.session.bulk_update_mappings()` → `db.session.execute(update(Model), dicts)`

```python
# Before
db.session.bulk_update_mappings(cls, [{'id': 1, 'col': 'val'}, ...])

# After
from sqlalchemy import update
db.session.execute(update(cls), [{'id': 1, 'col': 'val'}, ...])
db.session.commit()
```

> **Commit strategy:** Explicit `db.session.commit()` is shown here because
> `bulk_update_mappings` was often called in batch/background contexts. Inside a normal
> Flask request, Flask-SQLAlchemy commits automatically on request teardown — do **not**
> add an extra `commit()` in request-scoped model classmethods (see §3.3–3.5).

---

## 4. `.select_from(query)` → `.select_from(query.subquery())`

In SA 2.0, passing a `select()` statement directly to `.select_from()` requires
wrapping it in `.subquery()` first.

```python
# Before
db.session.execute(
    select([func.count()]).select_from(query)
).scalar()

# After
db.session.execute(
    select(func.count()).select_from(query.subquery())
).scalar_one()
```

### Existence check — belongs in a model classmethod, not in logic

```python
# ❌ Wrong — raw DB call in a logic function
# <pkg>/logic/some_resource.py
if db.session.execute(
    select(func.count()).select_from(
        SomeModel.filter_active()
        .where(
            SomeModel.period_id == period_id,
            SomeModel.contract_id.in_(contracts),
        )
        .subquery()
    )
).scalar_one():
    raise LogicError(error.ERROR_RESOURCE_ALREADY_EXISTS)

# ✅ Correct — model owns the DB call, logic calls the model
# <pkg>/models/some_resource.py
@classmethod
def exists_for_contracts_and_period(cls, contract_ids, period_id) -> bool:
    return bool(
        db.session.execute(
            select(func.count()).select_from(
                cls.filter_active()
                .where(
                    cls.period_id == period_id,
                    cls.contract_id.in_(contract_ids),
                )
                .subquery()
            )
        ).scalar_one()
    )

# <pkg>/logic/some_resource.py
if SomeModel.exists_for_contracts_and_period(
    contract_ids=contracts,
    period_id=period_id,
):
    raise LogicError(error.ERROR_RESOURCE_ALREADY_EXISTS)
```

---

## 5. Result Handling Changes

| Old SA 1.x pattern | SA 2.0 replacement | Notes |
|---|---|---|
| `.all()` on `Query` | `db.session.execute(stmt).scalars().all()` | Returns ORM objects |
| `.first()` on `Query` | `db.session.execute(stmt).scalars().first()` | Returns ORM object or `None` |
| `.one()` on `Query` | `db.session.execute(stmt).scalars().one()` | Raises if not exactly 1 |
| `.count()` on `Query` | `db.session.execute(select(func.count()).select_from(stmt.subquery())).scalar_one()` | Integer count |
| `.fetchall()` returning `[(obj,)]` tuples | `.scalars().all()` | Unwraps the tuple |
| `db.session.query(col1, col2).all()` | `db.session.execute(select(col1, col2)).mappings().all()` | Dict-like `RowMapping` |
| `query.one()` for aggregate row | `db.session.execute(stmt).one()` | Returns a `Row` namedtuple |

---

## 6. `filter()` → `where()` on `select()` statements

When building queries using the Core `select()` API (not `Query`), prefer `.where()`:

```python
# Before (Query API)
cls.query.filter(cls.deleted_at.is_(None))
# After (Core select)
select(cls).where(cls.deleted_at.is_(None))
```

> `filter()` still works as an alias on `select()` in SA 2.0 — but `.where()` is the
> idiomatic style and avoids confusion.

---

## 7. `base_list_query()` / Multi-Column Select Methods and Pagination

When a classmethod builds a multi-column `select()` (not returning full ORM instances),
the **execution and pagination must live on the model or in the repository**, not in logic.

```python
# ❌ Wrong — logic directly executes and paginates a raw select statement
# <pkg>/logic/some_resource.py
def get_items(limit, offset):
    query = SomeModel.base_list_query()
    items = db.session.execute(query.offset(offset).limit(limit)).mappings().all()
    count = db.session.execute(
        select(func.count()).select_from(query.subquery())
    ).scalar_one()
    return Items(items, count)

# ✅ Correct — model exposes a paginated method, logic just calls it
# <pkg>/models/some_resource.py
@classmethod
def get_paginated(cls, limit: int, offset: int):
    """Return (items, total_count) for the resource list."""
    stmt = cls.base_list_query()
    items = db.session.execute(stmt.offset(offset).limit(limit)).mappings().all()
    count = db.session.execute(
        select(func.count()).select_from(stmt.subquery())
    ).scalar_one()
    return items, count

# <pkg>/logic/some_resource.py
def get_items(limit: int, offset: int) -> Items:
    items, count = SomeModel.get_paginated(limit, offset)
    return Items(items, count)
```

> **Result type:** Multi-column selects (not returning full ORM objects) use
> `.mappings().all()` to return `RowMapping` objects — dict-like, accessed by column
> label name.

---

## 8. Type Annotation Change for Aggregate Row Return

When a classmethod returns an aggregate row (not an ORM object), update the return
type hint from `dict[str, ...]` to `Row`:

```python
from sqlalchemy import Row

# Before
@classmethod
def get_overview_by_event_id(cls, event_id: int) -> dict[str, Decimal]:
    ...

# After
@classmethod
def get_overview_by_event_id(cls, event_id: int) -> Row:
    ...
```

---

## 9. Tests — Updating Unit and Functional Tests

### 9.1 Replace `Model.query.all()` in tests
```python
# Before
SomeModel.query.all()

# After
from abacus_common_logic.models.base import db
from sqlalchemy import select

db.session.execute(select(SomeModel)).scalars().all()
```

### 9.2 Replace `.count()` assertions
```python
# Before
assert SomeModel.query.filter(SomeModel.id.in_(ids)).count() == 3

# After
from sqlalchemy import func, select

assert (
    db.session.execute(
        select(func.count()).select_from(
            select(SomeModel).where(SomeModel.id.in_(ids)).subquery()
        )
    ).scalar_one()
    == 3
)
```

### 9.3 `.fetchall()` row unpacking in tests

Previously `db.session.execute(stmt).fetchall()` returned `[(obj,)]` tuples.
With `.scalars().all()` it returns unwrapped objects directly:

```python
# Before
results = db.session.execute(stmt).fetchall()
items = [row[0] for row in results]

# After
items = db.session.execute(stmt).scalars().all()
```

---

## 10. Quick Reference — Complete Pattern Mapping

> All patterns in this table belong in **model or repository files only**.
> Logic files call named model/repository methods — they never use these constructs directly.

| SA 1.x pattern | SA 2.0 replacement |
|---|---|
| `db.engine.execute(sql)` | `with db.engine.connect() as c: c.execute(text(sql)); c.commit()` |
| `select([col1, col2])` | `select(col1, col2)` |
| `select(['*'])` | `select(literal_column('1'))` |
| `Model.query.filter(...).all()` | `db.session.execute(select(Model).where(...)).scalars().all()` |
| `Model.query.filter(...).first()` | `db.session.execute(select(Model).where(...)).scalars().first()` |
| `Model.query.filter_by(x=y).first()` | `db.session.execute(select(Model).where(Model.x == y)).scalars().first()` |
| `Model.query.filter(...).count()` | `db.session.execute(select(func.count()).select_from(select(Model).where(...).subquery())).scalar_one()` |
| `Model.query.filter_by(x=y).all()` | `db.session.execute(select(Model).where(Model.x == y)).scalars().all()` |
| `db.session.query(col1, col2).filter(...).all()` | `db.session.execute(select(col1, col2).where(...)).mappings().all()` |
| `db.session.query(Model).filter(...).all()` | `db.session.execute(select(Model).where(...)).scalars().all()` |
| `query.filter(...)` (on Query) | `stmt.where(...)` (on select statement) |
| `.query.filter(...).update({cls.col: val}, synchronize_session=False)` | `db.session.execute(update(Model).where(...).values(col=val).execution_options(synchronize_session=False))` |
| `.query.filter(...).delete(synchronize_session=False)` | `db.session.execute(delete(Model).where(...).execution_options(synchronize_session=False))` |
| `db.session.bulk_save_objects(objs)` | `db.session.add_all(objs)` (in model/repository only) |
| `db.session.bulk_update_mappings(Model, dicts)` | `db.session.execute(update(Model), dicts)` |
| `select([func.count()]).select_from(query)` | `select(func.count()).select_from(query.subquery())` |
| `execute(stmt).fetchall()` returning `[(obj,)]` | `execute(stmt).scalars().all()` |
| `execute(stmt).fetchall()` returning column rows | `execute(stmt).mappings().all()` |

---

## 11. Migration Execution Order

Follow these steps in order when migrating a service from SA 1.x to SA 2.0:

```
1.  Create a feature branch.

2.  [§1]   Bump pyproject.toml:
           - SQLAlchemy~=2.0.50
           - Flask-SQLAlchemy~=3.1.1
           - abacus-common-logic~=5.15.0

3.  Run:   uv sync && uv lock

4.  [§2.1] Add the _exec_sql() helper and replace db.engine.execute() in:
           - tests/conftest.py
           - tests/integration/conftest.py  (if it exists)

5.  [§2.2] Remove list brackets from all select([...]) calls.
           Search: grep -rn "select(\[" <pkg>/

6.  Start the service / run the test suite — note remaining failures.

7.  [§3.1] Replace Model.query usages in <pkg>/models/ and <pkg>/repository/.
           Search: grep -rn "\.query\." <pkg>/models/ <pkg>/repository/

8.  [§3.2] Replace db.session.query() usages in <pkg>/models/ and <pkg>/repository/.
           Search: grep -rn "db\.session\.query" <pkg>/models/ <pkg>/repository/

9.  [§3.3] Replace .query.filter().update() usages.
           Search: grep -rn "\.update(" <pkg>/models/ <pkg>/repository/

10. [§3.4] Replace .query.filter().delete() usages.
           Search: grep -rn "\.delete(" <pkg>/models/ <pkg>/repository/

11. [§3.5] Replace bulk_save_objects() with add_all() — ensure the call is in
           model/repository, not logic.
           Search: grep -rn "bulk_save_objects" <pkg>/

12. [§3.6] Replace bulk_update_mappings() with execute(update()).
           Search: grep -rn "bulk_update_mappings" <pkg>/

13. [§4]   Fix .select_from(query) → .select_from(query.subquery()).
           Search: grep -rn "select_from(" <pkg>/

14. [§5]   Fix result unpacking (.fetchall() → .scalars().all() or .mappings().all()).
           Search: grep -rn "\.fetchall()" <pkg>/

15. [§0]   Audit ALL logic files for any remaining db.session.* / sqlalchemy imports.
           Move every DB call into the model or repository layer.
           Search: grep -rn "db\.session\|from sqlalchemy" <pkg>/logic/

16. [§9]   Update unit and functional tests.
           Search: grep -rn "\.query\.\|db\.session\.query\|\.fetchall()" tests/

17. Run full test suite — expect green.
```

---

## 12. Audit Checklist (per file)

Use this checklist when reviewing any model, repository, logic, or test file during
migration:

### Model / Repository file
- [ ] No `select([...])` with list brackets
- [ ] No `Model.query.*`
- [ ] No `db.session.query(...)`
- [ ] `.query.filter(...).update(...)` replaced with `db.session.execute(update(...)...)`
- [ ] `.query.filter(...).delete(...)` replaced with `db.session.execute(delete(...)...)`
- [ ] `bulk_save_objects` replaced with `add_all`
- [ ] `bulk_update_mappings` replaced with `execute(update(Model), dicts)`
- [ ] `.select_from(query)` wrapped as `.select_from(query.subquery())`
- [ ] `.fetchall()` replaced with `.scalars().all()` or `.mappings().all()`
- [ ] Aggregate return type hints updated to `Row` where appropriate

### Logic file
- [ ] No `db.session.*` calls
- [ ] No `from sqlalchemy import ...`
- [ ] All DB operations delegated to model classmethods or repository functions

### Test file
- [ ] `Model.query.all()` replaced with `db.session.execute(select(Model)).scalars().all()`
- [ ] `.count()` assertions replaced (see §9.2)
- [ ] `.fetchall()` row unpacking updated (see §9.3)
- [ ] `db.engine.execute(sql)` replaced with `_exec_sql(sql)` helper (conftest files)