sqlalchemy 2.0.51: JSON

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

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:03a94c218b05b562b99beba5fe2bb0f90586fd32955c8ef239a8be74a335bd87","contract":["sqlalchemy.JSON maps complex nested dictionary and list structures with automatic serialization and deserialization","sqlalchemy.JSON supports JSON path element extraction and filtering using subscript expressions with type coercions like as_string and as_integer","sqlalchemy.JSON with none_as_null=True distinguishes between SQL NULL and JSON null literal values in database storage and queries","reassigning a modified dictionary to a sqlalchemy.JSON column triggers Session change tracking and persists updates"],"goal":"verify sqlalchemy.JSON in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.JSON"]},"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.JSON"],"verifierAdapter":"python@1"}

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

"""SQLAlchemy 2.0 JSON column type verification and usage sample."""

from typing import Any, Dict, List, Optional
from sqlalchemy import JSON, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class Document(Base):
    __tablename__ = "documents"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(100), nullable=False)
    payload: Mapped[Dict[str, Any]] = mapped_column(JSON, nullable=False)
    tags: Mapped[Optional[List[str]]] = mapped_column(JSON, nullable=True)
    metadata_json: Mapped[Optional[Any]] = mapped_column(JSON(none_as_null=True), nullable=True)

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


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


def insert_document(
    session: Session,
    title: str,
    payload: Dict[str, Any],
    tags: Optional[List[str]] = None,
    metadata_json: Optional[Any] = None,
) -> Document:
    """Insert a document with JSON payloads."""
    doc = Document(
        title=title,
        payload=payload,
        tags=tags,
        metadata_json=metadata_json,
    )
    session.add(doc)
    session.commit()
    session.refresh(doc)
    return doc


def find_by_json_key_string(session: Session, key: str, value: str) -> List[Document]:
    """Query documents filtering by JSON string field using subscript and as_string."""
    stmt = select(Document).where(Document.payload[key].as_string() == value).order_by(Document.id)
    return list(session.scalars(stmt).all())


def find_by_nested_json_integer(session: Session, parent_key: str, child_key: str, value: int) -> List[Document]:
    """Query documents filtering by nested JSON integer field using chained subscript and as_integer."""
    stmt = (
        select(Document)
        .where(Document.payload[parent_key][child_key].as_integer() == value)
        .order_by(Document.id)
    )
    return list(session.scalars(stmt).all())


if __name__ == "__main__":
    engine = create_database()
    with Session(engine) as session:
        doc = insert_document(
            session,
            "Example Doc",
            {"status": "active", "metrics": {"count": 10}},
            tags=["sample", "json"],
        )
        print(f"Created document: {doc}")
        results = find_by_json_key_string(session, "status", "active")
        print(f"Found {len(results)} document(s)")

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

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

import os
import sys

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy import JSON, select
from sqlalchemy.orm import Session
from sample import (
    Base,
    Document,
    create_database,
    insert_document,
    find_by_json_key_string,
    find_by_nested_json_integer,
)


def test_json_mapping_and_roundtrip() -> None:
    """Assertion 1: sqlalchemy.JSON maps complex nested dictionary and list structures with automatic serialization and deserialization."""
    assert issubclass(Document, Base), "Document must inherit from Base"
    assert hasattr(Document, "__table__"), "Document must have a mapped __table__"
    assert isinstance(Document.__table__.columns["payload"].type, JSON), "payload column must be of type JSON"
    assert isinstance(Document.__table__.columns["tags"].type, JSON), "tags column must be of type JSON"

    engine = create_database()
    with Session(engine) as session:
        complex_payload = {
            "version": 2,
            "config": {
                "active": True,
                "timeout": 30,
                "retries": [1, 2, 5],
            },
            "flags": {"experimental": False, "debug": None},
        }
        tags_list = ["alpha", "beta", "production"]

        doc = insert_document(session, "Doc 1", complex_payload, tags=tags_list)
        assert doc.id is not None, "Document must have generated primary key ID"

        # Query back in a new session / refresh to ensure fresh deserialization from database
        session.expire_all()
        fetched = session.scalars(select(Document).where(Document.id == doc.id)).one()

        assert fetched.payload == complex_payload, "Fetched payload must match inserted dictionary structure"
        assert fetched.payload["config"]["active"] is True, "Boolean field in JSON must roundtrip as bool"
        assert fetched.payload["config"]["retries"] == [1, 2, 5], "List in JSON payload must roundtrip as list"
        assert fetched.tags == ["alpha", "beta", "production"], "tags JSON column must roundtrip list"


def test_json_path_subscript_querying() -> None:
    """Assertion 2: sqlalchemy.JSON supports JSON path element extraction and filtering using subscript expressions with type coercions like as_string and as_integer."""
    engine = create_database()
    with Session(engine) as session:
        doc_a = insert_document(
            session,
            "Service A",
            {"service_name": "auth", "metrics": {"port": 8080, "workers": 4}},
        )
        doc_b = insert_document(
            session,
            "Service B",
            {"service_name": "billing", "metrics": {"port": 8081, "workers": 2}},
        )
        doc_c = insert_document(
            session,
            "Service C",
            {"service_name": "auth", "metrics": {"port": 9090, "workers": 8}},
        )

        # Filter top-level JSON key using as_string
        auth_services = find_by_json_key_string(session, "service_name", "auth")
        auth_ids = [s.id for s in auth_services if s.id in (doc_a.id, doc_b.id, doc_c.id)]
        assert len(auth_ids) == 2, f"Expected 2 auth services, got {len(auth_ids)}"
        assert auth_ids == [doc_a.id, doc_c.id]

        # Filter nested JSON key using as_integer
        port_8080 = find_by_nested_json_integer(session, "metrics", "port", 8080)
        port_8080_ids = [s.id for s in port_8080 if s.id in (doc_a.id, doc_b.id, doc_c.id)]
        assert port_8080_ids == [doc_a.id], f"Expected doc_a id, got {port_8080_ids}"

        port_8081 = find_by_nested_json_integer(session, "metrics", "port", 8081)
        port_8081_ids = [s.id for s in port_8081 if s.id in (doc_a.id, doc_b.id, doc_c.id)]
        assert port_8081_ids == [doc_b.id], f"Expected doc_b id, got {port_8081_ids}"


def test_sql_null_vs_json_null() -> None:
    """Assertion 3: sqlalchemy.JSON with none_as_null=True distinguishes between SQL NULL and JSON null literal values in database storage and queries."""
    engine = create_database()
    with Session(engine) as session:
        # SQL NULL for metadata_json
        doc_sql_null = insert_document(
            session,
            "Doc SQL NULL",
            {"key": "value"},
            metadata_json=None,
        )

        # JSON null literal in metadata_json
        doc_json_null = insert_document(
            session,
            "Doc JSON NULL",
            {"key": "value"},
            metadata_json=JSON.NULL,
        )

        session.expire_all()

        # Query verifying SQL NULL column for metadata_json
        sql_null_docs = session.scalars(
            select(Document).where(
                Document.id.in_([doc_sql_null.id, doc_json_null.id]),
                Document.metadata_json.is_(None),
            )
        ).all()
        assert len(sql_null_docs) == 1, "Only doc_sql_null has SQL NULL metadata_json"
        assert sql_null_docs[0].id == doc_sql_null.id

        # Query verifying JSON null is NOT SQL NULL in SQL queries
        json_null_docs = session.scalars(
            select(Document).where(
                Document.id.in_([doc_sql_null.id, doc_json_null.id]),
                Document.metadata_json.is_not(None),
            )
        ).all()
        assert len(json_null_docs) == 1, "doc_json_null has non-SQL-NULL JSON.NULL value"
        assert json_null_docs[0].id == doc_json_null.id

        # Verify JSON.NULL deserializes to None in Python
        fetched_json_null = session.scalars(
            select(Document).where(Document.id == doc_json_null.id)
        ).one()
        assert fetched_json_null.metadata_json is None, "JSON null evaluates to None in Python"


def test_json_mutation_and_persistence() -> None:
    """Assertion 4: reassigning a modified dictionary to a sqlalchemy.JSON column triggers Session change tracking and persists updates."""
    engine = create_database()
    with Session(engine) as session:
        doc = insert_document(
            session,
            "Mutable Doc",
            {"status": "pending", "items_count": 0},
        )
        doc_id = doc.id

        # Reassign modified dictionary to trigger tracking
        updated_payload = dict(doc.payload)
        updated_payload["status"] = "completed"
        updated_payload["items_count"] = 5
        doc.payload = updated_payload

        session.commit()

        # In a separate query, verify changes persisted
        session.expire_all()
        reloaded = session.scalars(select(Document).where(Document.id == doc_id)).one()
        assert reloaded.payload["status"] == "completed", "Updated JSON field must persist"
        assert reloaded.payload["items_count"] == 5, "Updated JSON integer field must persist"


def main() -> None:
    test_json_mapping_and_roundtrip()
    test_json_path_subscript_querying()
    test_sql_null_vs_json_null()
    test_json_mutation_and_persistence()
    print("All sqlalchemy.JSON contract assertions passed successfully.")


if __name__ == "__main__":
    main()