sqlalchemy 2.0.51: Boolean

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.Boolean 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.Boolean

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:99fe81b0d78ded94661be64564b2a212ea0e9f01e856e332523eef2ea7036968","contract":["sqlalchemy.Boolean defines boolean column types with python_type bool and optional check constraint support","sqlalchemy.Boolean maps Python True, False, and None values in ORM mapped models and roundtrips them via Session","sqlalchemy.Boolean supports boolean filtering in select statements using is_, equality, and nullable checks"],"goal":"verify sqlalchemy.Boolean in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.Boolean"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"subject":"pkg:pypi/sqlalchemy@2.0.51","symbols":["sqlalchemy.Boolean"],"verifierAdapter":"python@1"}

greenlet==3.5.5
sqlalchemy==2.0.51
typing-extensions==4.16.0

"""SQLAlchemy 2.0 Boolean column type and query demonstration."""

from typing import Optional, List
from sqlalchemy import Boolean, Engine, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(100), nullable=False)
    # Boolean with default False and NOT NULL constraint
    is_completed: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
    # Boolean with explicit check constraint and nullable=True
    is_archived: Mapped[Optional[bool]] = mapped_column(
        Boolean(create_constraint=True, name="ck_task_archived"),
        default=None,
        nullable=True,
    )

    def __repr__(self) -> str:
        return f"<Task(id={self.id}, title='{self.title}', is_completed={self.is_completed}, is_archived={self.is_archived})>"


def create_database(db_url: str = "sqlite:///:memory:") -> Engine:
    """Create database engine and initialize tables."""
    engine = create_engine(db_url, echo=False)
    Base.metadata.create_all(engine)
    return engine


def add_task(
    session: Session,
    title: str,
    is_completed: bool = False,
    is_archived: Optional[bool] = None,
) -> Task:
    """Add a new task entity."""
    task = Task(title=title, is_completed=is_completed, is_archived=is_archived)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


def set_task_completed(session: Session, task_id: int, completed: bool) -> Optional[Task]:
    """Update task completion status."""
    stmt = select(Task).where(Task.id == task_id)
    task = session.scalars(stmt).one_or_none()
    if task is not None:
        task.is_completed = completed
        session.commit()
        session.refresh(task)
    return task


def get_tasks_by_completion(session: Session, completed: bool) -> List[Task]:
    """Query tasks filtered by boolean completion flag."""
    stmt = select(Task).where(Task.is_completed.is_(completed)).order_by(Task.id)
    return list(session.scalars(stmt).all())


def get_unarchived_tasks(session: Session) -> List[Task]:
    """Query tasks where is_archived is None or False."""
    stmt = select(Task).where(
        (Task.is_archived.is_(None)) | (Task.is_archived.is_(False))
    ).order_by(Task.id)
    return list(session.scalars(stmt).all())

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

"""Contract verification test for sqlalchemy.Boolean in SQLAlchemy 2.0."""

import os
import sys

# Add sample root directory to sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy import Boolean, select
from sqlalchemy.orm import Session
from sample import (
    Base,
    Task,
    create_database,
    add_task,
    set_task_completed,
    get_tasks_by_completion,
    get_unarchived_tasks,
)


def test_boolean_column_metadata() -> None:
    """Assertion 1: sqlalchemy.Boolean defines boolean column types with python_type bool and optional check constraint support."""
    columns = Task.__table__.columns

    assert "is_completed" in columns, "Column 'is_completed' must be present"
    assert isinstance(columns["is_completed"].type, Boolean), "Column type must be Boolean"
    assert columns["is_completed"].type.python_type is bool, "python_type must be bool"
    assert columns["is_completed"].nullable is False, "is_completed must be non-nullable"

    assert "is_archived" in columns, "Column 'is_archived' must be present"
    assert isinstance(columns["is_archived"].type, Boolean), "Column type must be Boolean"
    assert columns["is_archived"].type.python_type is bool, "python_type must be bool"
    assert columns["is_archived"].nullable is True, "is_archived must be nullable"
    assert columns["is_archived"].type.create_constraint is True, "create_constraint must be True"
    assert columns["is_archived"].type.name == "ck_task_archived", "Constraint name must match"


def test_boolean_persistence_and_roundtrip() -> None:
    """Assertion 2: sqlalchemy.Boolean maps Python True, False, and None values in ORM mapped models and roundtrips them via Session."""
    engine = create_database()
    with Session(engine) as session:
        # Create records with distinct boolean values
        t1 = add_task(session, "Write docs", is_completed=False, is_archived=None)
        t2 = add_task(session, "Run tests", is_completed=True, is_archived=False)
        t3 = add_task(session, "Legacy task", is_completed=True, is_archived=True)

        assert t1.id is not None
        assert t2.id is not None
        assert t3.id is not None

        # Verify values persisted and returned as exact Python booleans and None
        q1 = session.get(Task, t1.id)
        assert q1 is not None
        assert q1.is_completed is False
        assert q1.is_archived is None

        q2 = session.get(Task, t2.id)
        assert q2 is not None
        assert q2.is_completed is True
        assert q2.is_archived is False

        q3 = session.get(Task, t3.id)
        assert q3 is not None
        assert q3.is_completed is True
        assert q3.is_archived is True

        # Update boolean field and verify roundtrip
        updated = set_task_completed(session, t1.id, True)
        assert updated is not None
        assert updated.is_completed is True

        refreshed = session.get(Task, t1.id)
        assert refreshed is not None
        assert refreshed.is_completed is True


def test_boolean_query_filtering() -> None:
    """Assertion 3: sqlalchemy.Boolean supports boolean filtering in select statements using is_, equality, and nullable checks."""
    engine = create_database()
    with Session(engine) as session:
        add_task(session, "Task A", is_completed=False, is_archived=None)
        add_task(session, "Task B", is_completed=True, is_archived=False)
        add_task(session, "Task C", is_completed=False, is_archived=True)
        add_task(session, "Task D", is_completed=True, is_archived=None)

        # Filter by is_completed == True
        completed = get_tasks_by_completion(session, True)
        assert len(completed) == 2, f"Expected 2 completed tasks, got {len(completed)}"
        assert {t.title for t in completed} == {"Task B", "Task D"}

        # Filter by is_completed == False
        pending = get_tasks_by_completion(session, False)
        assert len(pending) == 2, f"Expected 2 pending tasks, got {len(pending)}"
        assert {t.title for t in pending} == {"Task A", "Task C"}

        # Filter unarchived (None or False)
        unarchived = get_unarchived_tasks(session)
        assert len(unarchived) == 3, f"Expected 3 unarchived tasks, got {len(unarchived)}"
        assert {t.title for t in unarchived} == {"Task A", "Task B", "Task D"}

        # Direct boolean expression in where clause: Task.is_completed == True
        direct_true = list(session.scalars(select(Task).where(Task.is_completed == True)).all())
        assert len(direct_true) == 2

        # Direct boolean expression: Task.is_archived.is_(None)
        archived_none = list(session.scalars(select(Task).where(Task.is_archived.is_(None))).all())
        assert len(archived_none) == 2
        assert {t.title for t in archived_none} == {"Task A", "Task D"}


def main() -> None:
    test_boolean_column_metadata()
    test_boolean_persistence_and_roundtrip()
    test_boolean_query_filtering()
    print("All sqlalchemy.Boolean contract assertions passed successfully.")


if __name__ == "__main__":
    main()