sqlalchemy 2.0.51: exc.IntegrityError

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.exc.IntegrityError 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.exc.IntegrityError

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:056b196130f9296a6d9d6c06773205686b0a4af0b8b65a104ffa1a459f55811c","contract":["assert unique constraint violation raises sqlalchemy.exc.IntegrityError with statement and params attributes","assert NOT NULL constraint violation raises sqlalchemy.exc.IntegrityError","assert foreign key constraint violation raises sqlalchemy.exc.IntegrityError when foreign keys are enabled","assert CHECK constraint violation raises sqlalchemy.exc.IntegrityError","assert ORM session commit with duplicate unique field raises sqlalchemy.exc.IntegrityError and allows session rollback","assert IntegrityError is a subclass of DatabaseError and SQLAlchemyError and wraps underlying DBAPI exception in orig attribute"],"goal":"verify sqlalchemy.exc.IntegrityError in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.exc.IntegrityError"]},"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.exc.IntegrityError"],"verifierAdapter":"python@1"}

sqlalchemy==2.0.51
typing-extensions==4.16.0

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

"""Demonstration of sqlalchemy.exc.IntegrityError scenarios in SQLAlchemy 2.0."""

from typing import Any, Dict, List, Optional, Tuple
from sqlalchemy import (
    CheckConstraint,
    Column,
    ForeignKey,
    Integer,
    MetaData,
    String,
    Table,
    create_engine,
    event,
    insert,
)
from sqlalchemy.engine import Engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class UserModel(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(100), nullable=False)


def create_sqlite_engine(enable_foreign_keys: bool = True) -> Engine:
    """Create an in-memory SQLite engine, optionally enforcing foreign keys."""
    engine = create_engine("sqlite:///:memory:", echo=False)
    if enable_foreign_keys:
        @event.listens_for(engine, "connect")
        def set_sqlite_pragma(dbapi_connection, connection_record):
            cursor = dbapi_connection.cursor()
            cursor.execute("PRAGMA foreign_keys=ON")
            cursor.close()
    return engine


def build_core_schema(metadata: MetaData) -> Tuple[Table, Table, Table]:
    """Define Core tables with unique, not null, check, and foreign key constraints."""
    departments = Table(
        "departments",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String(50), nullable=False, unique=True),
    )

    employees = Table(
        "employees",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("badge_num", String(20), nullable=False, unique=True),
        Column("salary", Integer, nullable=False),
        Column("dept_id", Integer, ForeignKey("departments.id"), nullable=False),
        CheckConstraint("salary >= 0", name="chk_positive_salary"),
    )

    items = Table(
        "items",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("sku", String(30), nullable=False, unique=True),
        Column("description", String(200), nullable=False),
    )

    return departments, employees, items


def trigger_core_unique_violation(engine: Engine, metadata: MetaData) -> Tuple[bool, Optional[IntegrityError]]:
    """Attempt inserting duplicate unique values using Core API to trigger IntegrityError."""
    metadata.create_all(engine)
    items = metadata.tables["items"]

    with engine.begin() as conn:
        conn.execute(insert(items).values(id=1, sku="SKU-100", description="Item 1"))

    try:
        with engine.begin() as conn:
            conn.execute(insert(items).values(id=2, sku="SKU-100", description="Duplicate SKU Item"))
        return False, None
    except IntegrityError as exc:
        return True, exc


def trigger_core_not_null_violation(engine: Engine, metadata: MetaData) -> Tuple[bool, Optional[IntegrityError]]:
    """Attempt inserting NULL into non-nullable column using Core API to trigger IntegrityError."""
    metadata.create_all(engine)
    items = metadata.tables["items"]

    try:
        with engine.begin() as conn:
            conn.execute(insert(items).values(id=10, sku=None, description="Missing SKU"))
        return False, None
    except IntegrityError as exc:
        return True, exc


def trigger_core_foreign_key_violation(engine: Engine, metadata: MetaData) -> Tuple[bool, Optional[IntegrityError]]:
    """Attempt inserting record referencing non-existent parent to trigger foreign key IntegrityError."""
    metadata.create_all(engine)
    employees = metadata.tables["employees"]

    try:
        with engine.begin() as conn:
            conn.execute(
                insert(employees).values(id=1, badge_num="BADGE-01", salary=50000, dept_id=999)
            )
        return False, None
    except IntegrityError as exc:
        return True, exc


def trigger_core_check_constraint_violation(engine: Engine, metadata: MetaData) -> Tuple[bool, Optional[IntegrityError]]:
    """Attempt inserting record violating CHECK constraint to trigger IntegrityError."""
    metadata.create_all(engine)
    departments = metadata.tables["departments"]
    employees = metadata.tables["employees"]

    with engine.begin() as conn:
        conn.execute(insert(departments).values(id=1, name="Engineering"))

    try:
        with engine.begin() as conn:
            conn.execute(
                insert(employees).values(id=2, badge_num="BADGE-02", salary=-500, dept_id=1)
            )
        return False, None
    except IntegrityError as exc:
        return True, exc


def trigger_orm_duplicate_flush(engine: Engine) -> Tuple[bool, Optional[IntegrityError]]:
    """Attempt committing duplicate ORM instances to trigger IntegrityError on flush/commit."""
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        user1 = UserModel(username="user_alpha", email="user_alpha@example.com")
        session.add(user1)
        session.commit()

    with Session(engine) as session:
        user2 = UserModel(username="user_alpha", email="duplicate@example.com")
        session.add(user2)
        try:
            session.commit()
            return False, None
        except IntegrityError as exc:
            session.rollback()
            return True, exc

"""Contract verification tests for sqlalchemy.exc.IntegrityError in pkg:pypi/sqlalchemy@2.0.51."""

import os
import sys

# Ensure repository root is on sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy import MetaData
from sqlalchemy.exc import DatabaseError, IntegrityError, SQLAlchemyError

from src.integrity_demo import (
    Base,
    UserModel,
    build_core_schema,
    create_sqlite_engine,
    trigger_core_check_constraint_violation,
    trigger_core_foreign_key_violation,
    trigger_core_not_null_violation,
    trigger_core_unique_violation,
    trigger_orm_duplicate_flush,
)


def test_unique_constraint_raises_integrity_error():
    """Verify unique constraint violation raises sqlalchemy.exc.IntegrityError."""
    engine = create_sqlite_engine()
    metadata = MetaData()
    build_core_schema(metadata)

    caught, exc = trigger_core_unique_violation(engine, metadata)
    assert caught is True, "Expected IntegrityError on duplicate unique key insert"
    assert isinstance(exc, IntegrityError)
    assert isinstance(exc, DatabaseError)
    assert isinstance(exc, SQLAlchemyError)
    assert exc.statement is not None
    assert exc.params is not None
    assert exc.orig is not None


def test_not_null_constraint_raises_integrity_error():
    """Verify NOT NULL constraint violation raises sqlalchemy.exc.IntegrityError."""
    engine = create_sqlite_engine()
    metadata = MetaData()
    build_core_schema(metadata)

    caught, exc = trigger_core_not_null_violation(engine, metadata)
    assert caught is True, "Expected IntegrityError on NULL insert into non-nullable column"
    assert isinstance(exc, IntegrityError)
    assert exc.statement is not None
    assert exc.orig is not None


def test_foreign_key_constraint_raises_integrity_error():
    """Verify foreign key constraint violation raises sqlalchemy.exc.IntegrityError."""
    engine = create_sqlite_engine(enable_foreign_keys=True)
    metadata = MetaData()
    build_core_schema(metadata)

    caught, exc = trigger_core_foreign_key_violation(engine, metadata)
    assert caught is True, "Expected IntegrityError when referencing missing foreign key"
    assert isinstance(exc, IntegrityError)
    assert exc.statement is not None
    assert exc.orig is not None


def test_check_constraint_raises_integrity_error():
    """Verify CHECK constraint violation raises sqlalchemy.exc.IntegrityError."""
    engine = create_sqlite_engine()
    metadata = MetaData()
    build_core_schema(metadata)

    caught, exc = trigger_core_check_constraint_violation(engine, metadata)
    assert caught is True, "Expected IntegrityError when violating CHECK constraint"
    assert isinstance(exc, IntegrityError)
    assert exc.statement is not None
    assert exc.orig is not None


def test_orm_duplicate_flush_raises_integrity_error():
    """Verify ORM session commit with duplicate unique field raises sqlalchemy.exc.IntegrityError."""
    engine = create_sqlite_engine()
    caught, exc = trigger_orm_duplicate_flush(engine)
    assert caught is True, "Expected IntegrityError on committing duplicate ORM entity"
    assert isinstance(exc, IntegrityError)
    assert exc.statement is not None
    assert exc.orig is not None


def test_integrity_error_attributes_and_hierarchy():
    """Verify IntegrityError inherits from SQLAlchemyError and DatabaseError with orig and params attributes."""
    engine = create_sqlite_engine()
    metadata = MetaData()
    build_core_schema(metadata)

    _, exc = trigger_core_unique_violation(engine, metadata)
    assert exc is not None
    assert issubclass(IntegrityError, (DatabaseError, SQLAlchemyError, Exception))
    assert hasattr(exc, "statement")
    assert hasattr(exc, "params")
    assert hasattr(exc, "orig")
    assert hasattr(exc, "connection_invalidated")


def main():
    test_unique_constraint_raises_integrity_error()
    test_not_null_constraint_raises_integrity_error()
    test_foreign_key_constraint_raises_integrity_error()
    test_check_constraint_raises_integrity_error()
    test_orm_duplicate_flush_raises_integrity_error()
    test_integrity_error_attributes_and_hierarchy()
    print("All sqlalchemy.exc.IntegrityError contract tests passed successfully.")
    return 0


if __name__ == "__main__":
    sys.exit(main())