sqlalchemy 2.0.51: delete

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify sqlalchemy.delete in pkg:pypi/sqlalchemy@2.0.51
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/sqlalchemy@2.0.51
Demonstrate these symbols/APIs:
  - sqlalchemy.delete

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:c4a388c1c133c1386d89ca57ad9bd7c6b75cf2d909c93c3e0511e8888fb12722","contract":["delete construct generates DELETE SQL statement and compiles with parameters","delete with where clause removes matching rows in Core execution and reports rowcount","delete without where clause removes all rows from table","delete executes in ORM 2.0 Session via session.execute with model entity","delete with returning clause returns deleted column values upon execution"],"goal":"verify sqlalchemy.delete in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.delete"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"subject":"pkg:pypi/sqlalchemy@2.0.51","symbols":["sqlalchemy.delete"],"verifierAdapter":"python@1"}

[project]
name = "verify-sqlalchemy-delete"
version = "0.1.0"
dependencies = [
    "sqlalchemy==2.0.51",
]

sqlalchemy==2.0.51

{
  "schemaVersion": 1,
  "goal": "verify sqlalchemy.delete in pkg:pypi/sqlalchemy@2.0.51",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/sqlalchemy@2.0.51"
  ],
  "symbols": [
    "sqlalchemy.delete"
  ]
}

import sqlalchemy as sa
from sqlalchemy import (
    Column,
    Integer,
    MetaData,
    String,
    Table,
    create_engine,
    delete,
    insert,
    select,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


def test_delete_compilation():
    # Assertion 1: delete construct generates DELETE SQL statement and compiles with parameters
    metadata = MetaData()
    users = Table(
        "users",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String(50)),
    )
    stmt = delete(users).where(users.c.id == 42)
    compiled = str(stmt)
    assert "DELETE FROM users" in compiled, f"Unexpected compilation: {compiled}"
    assert "users.id = :id_1" in compiled or "users.id =" in compiled, f"Missing where clause: {compiled}"
    assert stmt.is_delete, "Statement is_delete attribute should be True"


def test_delete_core_with_where():
    # Assertion 2: delete with where clause removes matching rows in Core execution and reports rowcount
    engine = create_engine("sqlite:///:memory:")
    metadata = MetaData()
    items = Table(
        "items",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("category", String(20)),
        Column("value", Integer),
    )
    metadata.create_all(engine)

    with engine.connect() as conn:
        conn.execute(
            insert(items),
            [
                {"id": 1, "category": "A", "value": 10},
                {"id": 2, "category": "B", "value": 20},
                {"id": 3, "category": "A", "value": 30},
                {"id": 4, "category": "C", "value": 40},
            ],
        )
        conn.commit()

        # Delete category A items
        del_stmt = delete(items).where(items.c.category == "A")
        result = conn.execute(del_stmt)
        conn.commit()

        assert result.rowcount == 2, f"Expected 2 deleted rows, got {result.rowcount}"

        remaining = conn.execute(select(items.c.id).order_by(items.c.id)).scalars().all()
        assert remaining == [2, 4], f"Expected remaining ids [2, 4], got {remaining}"


def test_delete_all_without_where():
    # Assertion 3: delete without where clause removes all rows from table
    engine = create_engine("sqlite:///:memory:")
    metadata = MetaData()
    logs = Table(
        "logs",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("message", String(100)),
    )
    metadata.create_all(engine)

    with engine.connect() as conn:
        conn.execute(
            insert(logs),
            [{"id": 1, "message": "log1"}, {"id": 2, "message": "log2"}],
        )
        conn.commit()

        # Delete all rows
        result = conn.execute(delete(logs))
        conn.commit()

        assert result.rowcount == 2, f"Expected 2 rows deleted, got {result.rowcount}"

        count = conn.execute(select(sa.func.count()).select_from(logs)).scalar()
        assert count == 0, f"Expected 0 remaining rows, got {count}"


def test_delete_orm_session():
    # Assertion 4: delete executes in ORM 2.0 Session via session.execute with model entity
    class Base(DeclarativeBase):
        pass

    class User(Base):
        __tablename__ = "users_orm"
        id: Mapped[int] = mapped_column(primary_key=True)
        name: Mapped[str] = mapped_column(String(50))
        active: Mapped[bool] = mapped_column(default=True)

    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        session.add_all([
            User(id=1, name="Alice", active=True),
            User(id=2, name="Bob", active=False),
            User(id=3, name="Charlie", active=False),
        ])
        session.commit()

        # 2.0 style ORM delete statement
        stmt = delete(User).where(User.active.is_(False))
        res = session.execute(stmt)
        session.commit()

        assert res.rowcount == 2, f"Expected 2 rows deleted, got {res.rowcount}"

        remaining_names = session.scalars(select(User.name)).all()
        assert remaining_names == ["Alice"], f"Expected ['Alice'], got {remaining_names}"


def test_delete_returning():
    # Assertion 5: delete with returning clause returns deleted column values upon execution
    engine = create_engine("sqlite:///:memory:")
    metadata = MetaData()
    tasks = Table(
        "tasks",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("title", String(50)),
    )
    metadata.create_all(engine)

    with engine.connect() as conn:
        conn.execute(
            insert(tasks),
            [{"id": 1, "title": "task1"}, {"id": 2, "title": "task2"}],
        )
        conn.commit()

        stmt = delete(tasks).where(tasks.c.id == 1).returning(tasks.c.id, tasks.c.title)
        result = conn.execute(stmt)
        rows = result.fetchall()
        conn.commit()

        assert len(rows) == 1, f"Expected 1 returned row, got {len(rows)}"
        assert rows[0] == (1, "task1"), f"Expected (1, 'task1'), got {rows[0]}"


def main():
    test_delete_compilation()
    test_delete_core_with_where()
    test_delete_all_without_where()
    test_delete_orm_session()
    test_delete_returning()
    print("All contract assertions passed.")


if __name__ == "__main__":
    main()