sqlalchemy 2.0.51: event

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

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:ba410f4f24be5051064297f31cb7c71a1479311232ca34b65a07f217e8ba41f5","contract":["event.listen registers listener function on Engine target and event.contains confirms registration","event.remove unregisters listener function from Engine and raises InvalidRequestError when removing non-registered listener","event.listens_for decorator registers listener function on Engine target for before_cursor_execute","event.listen with named=True passes event arguments as keyword arguments to listener function","event.listen with retval=True allows listener function to transform executed SQL statement and parameters","event.listen and event.remove register and unregister ORM Session class-level event listeners"],"goal":"verify sqlalchemy.event in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.event"]},"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.event"],"verifierAdapter":"python@1"}

sqlalchemy==2.0.51
typing-extensions==4.16.0

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

"""Event management module demonstrating sqlalchemy.event core APIs."""

from typing import Any, Callable, Dict, List, Optional, Tuple
import sqlalchemy as sa
from sqlalchemy import Engine, event
from sqlalchemy.orm import Session


class QueryLogger:
    """Manages statement logging via sqlalchemy.event listeners on an Engine."""

    def __init__(self, engine: Engine) -> None:
        self.engine = engine
        self.logged_statements: List[str] = []

    def _on_before_cursor_execute(
        self,
        conn: Any,
        cursor: Any,
        statement: str,
        parameters: Any,
        context: Any,
        executemany: bool,
    ) -> None:
        self.logged_statements.append(statement)

    def attach(self) -> None:
        """Register the before_cursor_execute event listener."""
        if not self.is_attached():
            event.listen(self.engine, "before_cursor_execute", self._on_before_cursor_execute)

    def detach(self) -> None:
        """Remove the before_cursor_execute event listener."""
        if self.is_attached():
            event.remove(self.engine, "before_cursor_execute", self._on_before_cursor_execute)

    def is_attached(self) -> bool:
        """Check if listener is currently registered."""
        return event.contains(self.engine, "before_cursor_execute", self._on_before_cursor_execute)


class QueryRewriter:
    """Demonstrates retval=True listener to rewrite SQL statements before execution."""

    def __init__(self, engine: Engine, target_pattern: str, replacement_sql: str) -> None:
        self.engine = engine
        self.target_pattern = target_pattern
        self.replacement_sql = replacement_sql

    def _rewrite_listener(
        self,
        conn: Any,
        cursor: Any,
        statement: str,
        parameters: Any,
        context: Any,
        executemany: bool,
    ) -> Tuple[str, Any]:
        if self.target_pattern in statement:
            return self.replacement_sql, parameters
        return statement, parameters

    def attach(self) -> None:
        """Register with retval=True to allow transforming statement/parameters."""
        event.listen(self.engine, "before_cursor_execute", self._rewrite_listener, retval=True)

    def detach(self) -> None:
        """Remove the rewrite listener."""
        event.remove(self.engine, "before_cursor_execute", self._rewrite_listener)

    def is_attached(self) -> bool:
        """Check if listener is registered."""
        return event.contains(self.engine, "before_cursor_execute", self._rewrite_listener)


class NamedEventListener:
    """Demonstrates named=True listener receiving keyword arguments."""

    def __init__(self, engine: Engine) -> None:
        self.engine = engine
        self.received_kwargs: List[Dict[str, Any]] = []

    def _on_named_event(self, **kw: Any) -> None:
        self.received_kwargs.append(kw)

    def attach(self) -> None:
        """Register with named=True."""
        event.listen(self.engine, "before_cursor_execute", self._on_named_event, named=True)

    def detach(self) -> None:
        """Remove listener."""
        event.remove(self.engine, "before_cursor_execute", self._on_named_event)

    def is_attached(self) -> bool:
        return event.contains(self.engine, "before_cursor_execute", self._on_named_event)


def register_session_after_flush(callback: Callable[[Session, Any], None]) -> None:
    """Register an ORM Session class-level after_flush listener."""
    event.listen(Session, "after_flush", callback)


def remove_session_after_flush(callback: Callable[[Session, Any], None]) -> None:
    """Unregister an ORM Session class-level after_flush listener."""
    event.remove(Session, "after_flush", callback)


def contains_session_after_flush(callback: Callable[[Session, Any], None]) -> bool:
    """Check if callback is registered for Session after_flush."""
    return event.contains(Session, "after_flush", callback)

"""Contract verification test for sqlalchemy.event 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__))))

import sqlalchemy as sa
from sqlalchemy import create_engine, event, text
from sqlalchemy.exc import InvalidRequestError
from sqlalchemy.orm import Session

from src.event_manager import (
    NamedEventListener,
    QueryLogger,
    QueryRewriter,
    contains_session_after_flush,
    register_session_after_flush,
    remove_session_after_flush,
)


def test_event_listen_and_contains():
    """Verify event.listen registers listener and event.contains confirms it."""
    engine = create_engine("sqlite:///:memory:")
    logger = QueryLogger(engine)

    assert not logger.is_attached()
    logger.attach()
    assert logger.is_attached()
    assert event.contains(engine, "before_cursor_execute", logger._on_before_cursor_execute) is True

    with engine.connect() as conn:
        conn.execute(text("SELECT 100"))

    assert len(logger.logged_statements) == 1
    assert "SELECT 100" in logger.logged_statements[0]
    logger.detach()


def test_event_remove_and_invalid_request_error():
    """Verify event.remove unregisters listener and raises InvalidRequestError when not registered."""
    engine = create_engine("sqlite:///:memory:")
    captured = []

    def simple_listener(conn, cursor, statement, parameters, context, executemany):
        captured.append(statement)

    event.listen(engine, "before_cursor_execute", simple_listener)
    assert event.contains(engine, "before_cursor_execute", simple_listener) is True

    with engine.connect() as conn:
        conn.execute(text("SELECT 200"))
    assert len(captured) == 1

    event.remove(engine, "before_cursor_execute", simple_listener)
    assert event.contains(engine, "before_cursor_execute", simple_listener) is False

    with engine.connect() as conn:
        conn.execute(text("SELECT 201"))
    assert len(captured) == 1  # Not called after removal

    # Removing an already-removed listener must raise InvalidRequestError
    try:
        event.remove(engine, "before_cursor_execute", simple_listener)
        assert False, "Expected InvalidRequestError when removing unregistered listener"
    except InvalidRequestError:
        pass


def test_event_listens_for_decorator():
    """Verify @event.listens_for decorator registers event listener."""
    engine = create_engine("sqlite:///:memory:")
    decorated_calls = []

    @event.listens_for(engine, "before_cursor_execute")
    def on_cursor_exec(conn, cursor, statement, parameters, context, executemany):
        decorated_calls.append(statement)

    assert event.contains(engine, "before_cursor_execute", on_cursor_exec) is True

    with engine.connect() as conn:
        conn.execute(text("SELECT 300"))

    assert len(decorated_calls) == 1
    assert "SELECT 300" in decorated_calls[0]
    event.remove(engine, "before_cursor_execute", on_cursor_exec)


def test_event_listen_named():
    """Verify event.listen with named=True passes keyword arguments to listener."""
    engine = create_engine("sqlite:///:memory:")
    named_helper = NamedEventListener(engine)

    named_helper.attach()
    assert named_helper.is_attached()

    with engine.connect() as conn:
        conn.execute(text("SELECT 400"))

    assert len(named_helper.received_kwargs) == 1
    kwargs = named_helper.received_kwargs[0]
    assert "statement" in kwargs
    assert "SELECT 400" in kwargs["statement"]
    assert "parameters" in kwargs
    assert "context" in kwargs

    named_helper.detach()
    assert not named_helper.is_attached()


def test_event_listen_retval():
    """Verify event.listen with retval=True allows transforming executed SQL statement."""
    engine = create_engine("sqlite:///:memory:")
    rewriter = QueryRewriter(engine, target_pattern="ORIGINAL_QUERY", replacement_sql="SELECT 777 AS result")

    rewriter.attach()
    assert rewriter.is_attached()

    with engine.connect() as conn:
        val = conn.execute(text("SELECT ORIGINAL_QUERY")).scalar()

    assert val == 777

    rewriter.detach()
    assert not rewriter.is_attached()


def test_session_class_events():
    """Verify event.listen and event.remove work for ORM Session class-level events."""
    flush_calls = []

    def on_flush_hook(session, flush_context):
        flush_calls.append("flush_fired")

    assert not contains_session_after_flush(on_flush_hook)
    register_session_after_flush(on_flush_hook)
    assert contains_session_after_flush(on_flush_hook) is True

    remove_session_after_flush(on_flush_hook)
    assert contains_session_after_flush(on_flush_hook) is False


def main():
    test_event_listen_and_contains()
    test_event_remove_and_invalid_request_error()
    test_event_listens_for_decorator()
    test_event_listen_named()
    test_event_listen_retval()
    test_session_class_events()
    print("Contract verification for sqlalchemy.event passed successfully.")
    return 0


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