SQLAlchemy 2 guard against 1.x and legacy Query code

# AGENTS.md: SQLAlchemy 2 project (1.x and legacy Query guard)

This project uses **SQLAlchemy 2.x** (`sqlalchemy>=2.1,<3`) in **2.0 style**: typed `Mapped[]` models,
`select()` statements and explicit transactions. Most SQLAlchemy code you have seen is 1.x. Some of it was removed,
some still runs as the "legacy" Query API, and one change loses writes without an error.

## Version facts
- Check the pin first: `uv pip show sqlalchemy`.
- Removed in 2.0: `engine.execute()`, plain SQL strings in `execute()` (use `text()`), `select([...])`,
  `MetaData(bind=)`, `autocommit=True`, and string keys on rows (`row["name"]`).
- No autocommit: `with engine.connect() as conn:` rolls back anything you didn't `conn.commit()`. Use
  `with engine.begin() as conn:` for writes. Sessions work the same way: `with Session(engine) as s, s.begin():`.
- `session.query(...)` still runs but is legacy. New code uses `session.scalars(select(...))` / `session.execute(...)`.
- If you're not sure of an API, check the 2.0 migration guide (https://docs.sqlalchemy.org/en/21/changelog/migration_20.html)
  instead of copying 1.x answers.

## Rules
- Models: `class Base(DeclarativeBase): pass`, then `name: Mapped[str] = mapped_column(String(80))`.
  `Mapped[str | None]` for nullable columns.
- Relationships: `relationship(back_populates=...)` on both sides. Use `lazy="raise"` or `selectinload` when N+1
  queries matter. Use `WriteOnlyMapped` for big collections, not `lazy="dynamic"`.
- Queries: `select(User).where(User.name == name)`, then `session.scalars(stmt).all()` or `.one_or_none()`.
  Use `session.get(User, pk)` for primary keys.
- Raw SQL only through `text("... :name")` with bound parameters. Never use f-strings.
- Writes happen inside `engine.begin()` or `session.begin()`, or are followed by an explicit commit.
- Don't upgrade or downgrade SQLAlchemy without asking the human.

## Before you say "done"
1. Run the `sqlalchemy-2-guard` scanner on the files you touched. It must report `0 error(s)`. Read every `warn`.
2. Run the tests with SQLAlchemy's own warnings as errors:
   `uv run pytest -q -W error::sqlalchemy.exc.SADeprecationWarning`.
3. For every write path you touched, say which test shows the row is really there afterwards.

# CLAUDE.md

@AGENTS.md

## Claude Code notes

- Skills from this loadout live in `.claude/skills/`: `sqlalchemy-2-guard`. Use them when their description matches the task.
- Follow AGENTS.md above; where this file and AGENTS.md disagree, AGENTS.md wins.

<!-- From SetupBay "SQLAlchemy 2 guard against 1.x and legacy Query code" v1.0.0 — https://setupbay.com/loadouts/sqlalchemy-2-style-guard/ -->

# GEMINI.md

@./AGENTS.md

## Gemini CLI notes

- Skills from this loadout live in `.agents/skills/`: `sqlalchemy-2-guard`. Use them when their description matches the task.
- Follow AGENTS.md above; where this file and AGENTS.md disagree, AGENTS.md wins.

<!-- From SetupBay "SQLAlchemy 2 guard against 1.x and legacy Query code" v1.0.0 — https://setupbay.com/loadouts/sqlalchemy-2-style-guard/ -->

---
description: SQLAlchemy 2.0 style only; no removed 1.x APIs or legacy Query code
globs: "**/*.py"
alwaysApply: false
---

This project uses SQLAlchemy 2.x in 2.0 style. There is no autocommit.

| Never (1.x / legacy) | Always (2.0 style) |
|---|---|
| `engine.execute("select ...")` | `with engine.begin() as conn: conn.execute(text("..."), params)` |
| `conn.execute("raw sql")` | `conn.execute(text("... :id"), {"id": 1})` |
| `select([User.id])` | `select(User.id)` |
| `session.query(User).filter_by(name=n).all()` | `session.scalars(select(User).where(User.name == n)).all()` |
| `session.query(User).get(1)` | `session.get(User, 1)` |
| `row["name"]` | `row.name` or `row._mapping["name"]` |
| `declarative_base()`, `id = Column(Integer, ...)` | `class Base(DeclarativeBase)`, `id: Mapped[int] = mapped_column(primary_key=True)` |
| `relationship(backref="x")`, `lazy="dynamic"` | `back_populates=` both sides, `WriteOnlyMapped` |
| `MetaData(bind=engine)`, `autocommit=True` | pass the engine; `with Session(engine) as s, s.begin():` |
| writes inside `engine.connect()` with no commit | `engine.begin()` |

- After editing, run the `sqlalchemy-2-guard` skill's scanner and the tests.

---
name: sqlalchemy-2-guard
description: Check Python for SQLAlchemy 1.x APIs removed in 2.0, legacy session.query code, untyped models and writes that are never committed, before saying a database change is done. Use after writing or editing SQLAlchemy models, queries or transactions, when SQLAlchemy raises ObjectNotExecutableError or ArgumentError, or when porting 1.x snippets.
---

# SQLAlchemy 2 guard

SQLAlchemy 2.0 came out in January 2023, and agents still write 1.x: `engine.execute("...")`, `select([...])`,
`row["name"]`, `session.query()`, `declarative_base()` with untyped `Column`s. The removed APIs fail at once.
`session.query()` keeps working as legacy code. The dangerous change is quiet: there is no autocommit, so a write
done inside `engine.connect()` without `commit()` is rolled back when the block ends, and no error is raised.

## 1. Confirm the version
- `uv run python -c "import sqlalchemy; print(sqlalchemy.__version__)"` should print `2.x`. If it prints `1.4`,
  ask before migrating. 1.4 with `SQLALCHEMY_WARN_20=1` is the stepping stone.

## 2. Scan
```bash
python3 .agents/skills/sqlalchemy-2-guard/scripts/sqlalchemy2_guard.py app tests
```
(Use the path the skill was installed to: `.claude/skills/...` for Claude Code, `.agents/skills/...` for the others.)

- Every `error` line is removed 1.x code or the legacy Query API. Fix it with the replacement the line gives.
  `references/migration-table.md` has before/after code.
- `connect-then-write` warnings: if the block writes, switch to `engine.begin()` or call `conn.commit()`. If it only
  reads, add `# guard: ok`. That comment silences warnings, never errors.
- `untyped-column` and `backref` warnings: convert the model to `Mapped[]` / `mapped_column` and `back_populates`.
  These conversions don't change the schema. Run `alembic check` (or compare `CREATE TABLE` output) to confirm.

## 3. Let SQLAlchemy check it
```bash
uv run pytest -q -W error::sqlalchemy.exc.SADeprecationWarning
```
- Legacy calls such as `Query.get()` warn `LegacyAPIWarning`, a subclass of `SADeprecationWarning`, so they fail here.
- Every write test should read the row back in a **new** session or connection. If it reads it back in the same
  one, an uncommitted write still looks fine.

## 4. Report
Say how many 1.x calls you replaced, which write paths you checked for commits and with which test, and that the
scanner and the tests were clean.

#!/usr/bin/env python3
"""Find SQLAlchemy 1.x code (and the legacy Query API) in a project on SQLAlchemy 2.x.

Usage: python3 sqlalchemy2_guard.py [PATH ...]        (default: the current folder)
       python3 sqlalchemy2_guard.py --list            (print every rule)

A "guard: ok" comment on a line silences its warnings (never its errors).
Prints one line per finding:  path:line: level [rule] what is wrong -> what to use
Exit 0 when nothing at "error" level is found, 1 when something is, 2 on bad usage.
Standard library only. Read-only: it never edits a file and makes no network calls.
"""
import fnmatch
import os
import re
import sys

# (rule id, level, regex, message, file patterns). Patterns run against each line with comments removed.
RULES = [
    ('engine-execute', 'error', '\\b(engine|eng|db_engine|self\\.engine)\\.execute\\(',
     'Engine.execute() was removed in 2.0 -> with engine.begin() as conn: conn.execute(text(...))', []),
    ('select-list', 'error', '\\bselect\\(\\s*\\[',
     'select([a, b]) was removed -> select(a, b)', []),
    ('string-sql', 'error', '\\.execute\\(\\s*[frbu]*[\\"\']',
     'executing a plain SQL string was removed -> conn.execute(text("..."), {"name": value})', []),
    ('metadata-bind', 'error', '\\bMetaData\\([^)]*\\bbind\\s*=',
     'MetaData(bind=...) was removed -> pass the engine or connection explicitly', []),
    ('autocommit', 'error', '\\bautocommit\\s*=\\s*True',
     'Session/sessionmaker(autocommit=True) was removed -> with Session(engine) as s, s.begin(): ...', []),
    ('row-string-key', 'error', '\\b(row|r|result_row|record)\\[\\s*[\\"\']',
     "rows are tuples: row['name'] fails -> row.name or row._mapping['name']", []),
    ('ext-declarative', 'error', 'sqlalchemy\\.ext\\.declarative\\s+import\\s+[^#]*declarative_base',
     'sqlalchemy.ext.declarative.declarative_base moved -> class Base(DeclarativeBase): pass', []),
    ('query-get', 'error', '\\.query\\([^)]*\\)\\.get\\(',
     'Query.get() is legacy -> session.get(Model, pk)', []),
    ('legacy-query', 'error', '\\b(session|db|s|sess|self\\.session|db\\.session)\\.query\\(',
     'session.query() is the legacy Query API (still runs) -> session.scalars(select(Model).where(...))', []),
    ('orm-relation', 'error', '\\borm\\.relation\\(|\\brelation\\(',
     'relation() -> relationship()', []),
    ('declarative-base-fn', 'warn', '\\bdeclarative_base\\(\\s*\\)',
     'declarative_base() works but 2.0 style is class Base(DeclarativeBase): pass, with Mapped[] annotations', []),
    ('untyped-column', 'warn', '^\\s+\\w+\\s*=\\s*(sa\\.)?Column\\(',
     'Column() in a mapped class -> name: Mapped[str] = mapped_column(...) so types reach the type checker', []),
    ('backref', 'warn', '\\bbackref\\s*=',
     'backref= -> relationship(back_populates=...) on both sides, so both attributes are visible', []),
    ('future-flag', 'warn', '\\bfuture\\s*=\\s*True',
     'future=True does nothing on 2.x; remove it', []),
    ('dynamic-loader', 'warn', 'lazy\\s*=\\s*[\\"\']dynamic[\\"\']',
     "lazy='dynamic' is legacy -> WriteOnlyMapped[...] with lazy='write_only'", []),
    ('connect-then-write', 'warn', 'with\\s+\\w+\\.connect\\(\\)\\s+as\\s+\\w+\\s*:',
     'engine.connect() does not commit: writes are rolled back unless you call conn.commit() -> use engine.begin() for writes', []),
]
FILES = ['*.py']
COMMENT = '#'
SKIP_DIRS = {".git", ".venv", "venv", "node_modules", "__pycache__", ".agents", ".claude", ".cursor", "dist", "build", "target", ".godot"}
COMPILED = [(rid, lvl, re.compile(rx), msg, globs) for rid, lvl, rx, msg, globs in RULES]


def strip_comment(line):
    """Drop a trailing comment, leaving string literals alone."""
    q = None
    i = 0
    while i < len(line):
        c = line[i]
        if q:
            if c == "\\":
                i += 2
                continue
            if c == q:
                q = None
        elif c in "\"'`":
            q = c
        elif line.startswith(COMMENT, i):
            return line[:i]
        i += 1
    return line


def wanted(name):
    return any(fnmatch.fnmatch(name, g) for g in FILES)


def walk(paths):
    for p in paths:
        if os.path.isfile(p):
            yield p
            continue
        for root, dirs, files in os.walk(p):
            dirs[:] = sorted(d for d in dirs if d not in SKIP_DIRS)
            for f in sorted(files):
                if wanted(f):
                    yield os.path.join(root, f)


def scan(path):
    out = []
    name = os.path.basename(path)
    try:
        with open(path, encoding="utf-8", errors="replace") as fh:
            lines = fh.read().splitlines()
    except OSError as e:
        return [(path, 0, "error", "unreadable", str(e))]
    for n, raw in enumerate(lines, 1):
        code = strip_comment(raw)
        if not code.strip():
            continue
        accepted = "guard: ok" in raw  # a person accepted the warnings on this line
        for rid, lvl, rx, msg, globs in COMPILED:
            if accepted and lvl == "warn":
                continue
            if globs and not any(fnmatch.fnmatch(name, g) for g in globs):
                continue
            if rx.search(code):
                out.append((path, n, lvl, rid, msg))
    return out


def main(argv):
    if "--list" in argv:
        for rid, lvl, _, msg, globs in RULES:
            print(f"{lvl:5} [{rid}] {msg}" + (f"  (only {', '.join(globs)})" if globs else ""))
        return 0
    if any(a.startswith("-") for a in argv):
        print(__doc__)
        return 2
    paths = argv or ["."]
    for p in paths:
        if not os.path.exists(p):
            print(f"no such path: {p}", file=sys.stderr)
            return 2
    files = list(walk(paths))
    findings = []
    for f in files:
        findings.extend(scan(f))
    for path, n, lvl, rid, msg in findings:
        print(f"{os.path.relpath(path)}:{n}: {lvl} [{rid}] {msg}")
    errors = sum(1 for x in findings if x[2] == "error")
    warns = sum(1 for x in findings if x[2] == "warn")
    print(f"sqlalchemy2_guard: {len(files)} file(s), {errors} error(s), {warns} warning(s)")
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

# SQLAlchemy 1.x → 2.0 style table

From the SQLAlchemy 2.0 migration guide (https://docs.sqlalchemy.org/en/21/changelog/migration_20.html) and the
ORM querying guide. Checked against SQLAlchemy 2.1.6.

## Removed in 2.0 (raises)

| 1.x | 2.x | Error on 2.1.6 |
|---|---|---|
| `engine.execute("select 1")` | `with engine.connect() as c: c.execute(text("select 1"))` | `AttributeError` |
| `conn.execute("select 1")` | `conn.execute(text("select 1"))` | `ObjectNotExecutableError` |
| `select([User.id, User.name])` | `select(User.id, User.name)` | `ArgumentError` |
| `MetaData(bind=engine)` | `MetaData()`; pass the engine to `create_all(engine)` | `TypeError` |
| `Session(engine, autocommit=True)` | `with Session(engine) as s, s.begin(): ...` | `ArgumentError` |
| `row["name"]` | `row.name`, `row._mapping["name"]` | `TypeError` |

## Legacy (still runs, warns or is discouraged)

| 1.x | 2.0 style |
|---|---|
| `session.query(User).all()` | `session.scalars(select(User)).all()` |
| `session.query(User).filter_by(name=n).first()` | `session.scalars(select(User).filter_by(name=n)).first()` |
| `session.query(User).get(1)` (`LegacyAPIWarning`) | `session.get(User, 1)` |
| `session.query(func.count(User.id)).scalar()` | `session.scalar(select(func.count(User.id)))` |
| `from sqlalchemy.ext.declarative import declarative_base` (`MovedIn20Warning`) | `class Base(DeclarativeBase): pass` |
| `id = Column(Integer, primary_key=True)` | `id: Mapped[int] = mapped_column(primary_key=True)` |
| `name = Column(String, nullable=True)` | `name: Mapped[str \| None] = mapped_column(String(80))` |
| `relationship("Post", backref="author")` | `posts: Mapped[list["Post"]] = relationship(back_populates="author")` on both sides |
| `lazy="dynamic"` | `posts: WriteOnlyMapped["Post"] = relationship()` |
| `create_engine(url, future=True)` | `create_engine(url)` |

## Transactions: writes that vanish

| Code | What happens on 2.x | Fix |
|---|---|---|
| `with engine.connect() as c: c.execute(insert_stmt)` | rolled back at the end of the block | `with engine.begin() as c:` or `c.commit()` |
| `s = Session(engine); s.add(u)` and no commit | discarded when the session closes | `with Session(engine) as s, s.begin(): s.add(u)` |

# A SQLAlchemy 1.3/1.4 module, used to check the guard. Each line is removed, legacy or silently different in 2.x.
from sqlalchemy import Column, Integer, MetaData, String, create_engine, select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
engine = create_engine("sqlite:///app.db", future=True)
meta = MetaData(bind=engine)
Base = declarative_base()
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    posts = relationship("Post", backref="author", lazy="dynamic")
Session = sessionmaker(bind=engine, autocommit=True)
session = Session()
admins = session.query(User).filter_by(name="admin").all()
one = session.query(User).get(1)
rows = engine.execute("select id, name from users")
for row in rows:
    print(row["name"])
ids = select([User.id])
with engine.connect() as conn:
    conn.execute(text("update users set name = 'x'"))