docs: convert data_flow.rst code examples to executable tests

docs/examples/usage/usage_data_flow_1.py

@@ -0,0 +1,19 @@
+"""Example 1: Direct SQL Creation with positional and named parameters."""
+
+__all__ = ("test_direct_sql_creation",)
+
+
+def test_direct_sql_creation() -> None:
+    """Test creating SQL statements with different parameter styles."""
+    # start-example
+    from sqlspec import SQL
+
+    # Raw SQL string with positional parameters
+    sql = SQL("SELECT * FROM users WHERE id = ?", 1)
+
+    # Named parameters
+    sql = SQL("SELECT * FROM users WHERE email = :email", email="user@example.com")
+    # end-example
+
+    # Verify SQL objects were created
+    assert sql is not None

docs/examples/usage/usage_data_flow_10.py

@@ -0,0 +1,27 @@
+"""Example 10: SQLite Driver Execution implementation."""
+
+__all__ = ("test_sqlite_driver_pattern",)
+
+
+def test_sqlite_driver_pattern() -> None:
+    """Test SQLite driver execution pattern."""
+    # start-example
+    from typing import Any
+
+    from sqlspec.driver import ExecutionResult, SyncDriverAdapterBase
+
+    class SqliteDriver(SyncDriverAdapterBase):
+        def _execute_statement(self, cursor: Any, statement: Any) -> ExecutionResult:
+            sql, params = self._get_compiled_sql(statement)
+            cursor.execute(sql, params or ())
+            return self.create_execution_result(cursor)
+
+        def _execute_many(self, cursor: Any, statement: Any) -> ExecutionResult:
+            sql, params = self._get_compiled_sql(statement)
+            cursor.executemany(sql, params)
+            return self.create_execution_result(cursor)
+
+    # end-example
+
+    # Verify class was defined
+    assert SqliteDriver is not None

docs/examples/usage/usage_data_flow_11.py

@@ -0,0 +1,27 @@
+"""Example 11: SQLResult Object."""
+
+__all__ = ("test_sql_result_object",)
+
+
+def test_sql_result_object() -> None:
+    """Test accessing SQLResult object properties."""
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
+
+    db_manager = SQLSpec()
+    db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
+
+    # start-example
+    with db_manager.provide_session(db) as session:
+        result = session.execute("SELECT 'test' as col1, 'value' as col2")
+
+        # Access result data
+        result.data  # List of dictionaries
+        result.rows_affected  # Number of rows modified (INSERT/UPDATE/DELETE)
+        result.column_names  # Column names for SELECT
+        result.operation_type  # "SELECT", "INSERT", "UPDATE", "DELETE", "SCRIPT"
+    # end-example
+
+    # Verify result properties
+    assert result.data is not None
+    assert result.column_names is not None

docs/examples/usage/usage_data_flow_12.py

@@ -0,0 +1,35 @@
+"""Example 12: Convenience Methods."""
+
+__all__ = ("test_convenience_methods",)
+
+
+def test_convenience_methods() -> None:
+    """Test SQLResult convenience methods."""
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
+
+    db_manager = SQLSpec()
+    db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
+
+    with db_manager.provide_session(db) as session:
+        # Create a test table
+        session.execute("CREATE TABLE test (id INTEGER, name TEXT)")
+        session.execute("INSERT INTO test VALUES (1, 'Alice')")
+
+        # start-example
+        result = session.execute("SELECT * FROM test WHERE id = 1")
+
+        # Get exactly one row (raises if not exactly one)
+        user = result.one()
+
+        # Get one or None
+        user = result.one_or_none()
+
+        # Get scalar value (first column of first row)
+        result2 = session.execute("SELECT COUNT(*) FROM test")
+        count = result2.scalar()
+        # end-example
+
+        # Verify results
+        assert user is not None
+        assert count == 1

docs/examples/usage/usage_data_flow_13.py

@@ -0,0 +1,43 @@
+"""Example 13: Schema Mapping."""
+
+__all__ = ("test_schema_mapping",)
+
+
+def test_schema_mapping() -> None:
+    """Test mapping results to typed objects."""
+
+    # start-example
+    from pydantic import BaseModel
+
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
+
+    class User(BaseModel):
+        id: int
+        name: str
+        email: str
+        is_active: bool | None = True
+
+    db_manager = SQLSpec()
+    db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
+
+    with db_manager.provide_session(db) as session:
+        # Create test table
+        session.execute("CREATE TABLE users (id INTEGER, name TEXT, email TEXT, is_active INTEGER)")
+        session.execute("INSERT INTO users VALUES (1, 'Alice', 'alice@example.com', 1)")
+
+        # Execute query
+        result = session.execute("SELECT id, name, email, is_active FROM users")
+
+        # Map results to typed User instances
+        users: list[User] = result.all(schema_type=User)
+
+        # Or get single typed user
+        single_result = session.execute("SELECT id, name, email, is_active FROM users WHERE id = ?", 1)
+        user: User = single_result.one(schema_type=User)  # Type-safe!
+    # end-example
+
+    # Verify typed results
+    assert len(users) == 1
+    assert isinstance(user, User)
+    assert user.id == 1

docs/examples/usage/usage_data_flow_14.py

@@ -0,0 +1,18 @@
+"""Example 14: Multi-Tier Caching."""
+
+__all__ = ("test_multi_tier_caching",)
+
+
+def test_multi_tier_caching() -> None:
+    """Test cache types in SQLSpec."""
+    # start-example
+    from sqlglot import Expression
+
+    # Cache types and their purposes:
+    sql_cache: dict[str, str] = {}  # Compiled SQL strings
+    optimized_cache: dict[str, Expression] = {}  # Post-optimization AST
+    # end-example
+
+    # Verify cache dictionaries were created
+    assert isinstance(sql_cache, dict)
+    assert isinstance(optimized_cache, dict)

docs/examples/usage/usage_data_flow_15.py

@@ -0,0 +1,26 @@
+"""Example 15: Configuration-Driven Processing."""
+
+__all__ = ("test_configuration_driven_processing",)
+
+
+def test_configuration_driven_processing() -> None:
+    """Test StatementConfig for controlling pipeline behavior."""
+    # start-example
+    from sqlspec import ParameterStyle, ParameterStyleConfig, StatementConfig
+
+    config = StatementConfig(
+        dialect="postgres",
+        enable_parsing=True,  # AST generation
+        enable_validation=True,  # Security/performance checks
+        enable_transformations=True,  # AST transformations
+        enable_caching=True,  # Multi-tier caching
+        parameter_config=ParameterStyleConfig(
+            default_parameter_style=ParameterStyle.NUMERIC, has_native_list_expansion=False
+        ),
+    )
+    # end-example
+
+    # Verify config was created
+    assert config is not None
+    assert config.dialect == "postgres"
+    assert config.enable_parsing is True

docs/examples/usage/usage_data_flow_2.py

@@ -0,0 +1,16 @@
+"""Example 2: Using the Query Builder."""
+
+__all__ = ("test_query_builder",)
+
+
+def test_query_builder() -> None:
+    """Test building SQL programmatically."""
+    # start-example
+    from sqlspec import sql
+
+    # Build SQL programmatically
+    query = sql.select("id", "name", "email").from_("users").where("status = ?", "active")
+    # end-example
+
+    # Verify query object was created
+    assert query is not None

docs/examples/usage/usage_data_flow_3.py

@@ -0,0 +1,20 @@
+"""Example 3: From SQL Files."""
+
+from pathlib import Path
+
+__all__ = ("test_sql_file_loader",)
+
+
+def test_sql_file_loader() -> None:
+    """Test loading SQL from files."""
+    # start-example
+    from sqlspec.loader import SQLFileLoader
+
+    loader = SQLFileLoader()
+    queries_path = Path(__file__).resolve().parents[1] / "queries" / "users.sql"
+    loader.load_sql(queries_path)
+    sql = loader.get_sql("get_user_by_id")
+    # end-example
+
+    # Verify SQL object was created
+    assert sql is not None

docs/examples/usage/usage_data_flow_4.py

@@ -0,0 +1,19 @@
+"""Example 4: Parameter Extraction."""
+
+__all__ = ("test_parameter_extraction",)
+
+
+def test_parameter_extraction() -> None:
+    """Show how SQL captures positional and named parameters."""
+    from sqlspec import SQL
+
+    # start-example
+    positional = SQL("SELECT * FROM users WHERE id = ? AND status = ?", 1, "active")
+    positional_map = dict(enumerate(positional.positional_parameters))
+
+    named = SQL("SELECT * FROM users WHERE email = :email", email="user@example.com")
+    named_map = named.named_parameters
+    # end-example
+
+    assert positional_map == {0: 1, 1: "active"}
+    assert named_map == {"email": "user@example.com"}

docs/examples/usage/usage_data_flow_5.py

@@ -0,0 +1,16 @@
+"""Example 5: AST Generation with SQLGlot."""
+
+__all__ = ("test_ast_generation",)
+
+
+def test_ast_generation() -> None:
+    """Test parsing SQL into Abstract Syntax Tree."""
+    # start-example
+    import sqlglot
+
+    # Parse SQL into structured AST
+    expression = sqlglot.parse_one("SELECT * FROM users WHERE id = ?", dialect="sqlite")
+    # end-example
+
+    # Verify expression was created
+    assert expression is not None

docs/examples/usage/usage_data_flow_6.py

@@ -0,0 +1,18 @@
+"""Example 6: Compilation to target dialect."""
+
+__all__ = ("test_compilation",)
+
+
+def test_compilation() -> None:
+    """Test compiling AST to target SQL dialect."""
+    # start-example
+    import sqlglot
+
+    # Compile AST to target dialect
+    compiled_sql = sqlglot.parse_one("SELECT * FROM users WHERE id = ?", dialect="sqlite").sql(dialect="postgres")
+    # Result: "SELECT * FROM users WHERE id = $1"
+    # end-example
+
+    # Verify compilation produced a string
+    assert isinstance(compiled_sql, str)
+    assert "users" in compiled_sql

docs/examples/usage/usage_data_flow_7.py

@@ -0,0 +1,25 @@
+"""Example 7: Parameter Processing."""
+
+__all__ = ("test_parameter_processing",)
+
+
+def test_parameter_processing() -> None:
+    """Convert SQLite-style placeholders to PostgreSQL numeric parameters."""
+    from sqlspec import SQL, ParameterStyle, ParameterStyleConfig, StatementConfig
+
+    # start-example
+    statement_config = StatementConfig(
+        dialect="sqlite",
+        parameter_config=ParameterStyleConfig(
+            default_parameter_style=ParameterStyle.NUMERIC, has_native_list_expansion=False
+        ),
+    )
+
+    sql = SQL("SELECT * FROM users WHERE id = ? AND status = ?", 1, "active", statement_config=statement_config)
+
+    compiled_sql, execution_params = sql.compile()
+    # end-example
+
+    assert "$1" in compiled_sql
+    assert "$2" in compiled_sql
+    assert execution_params == [1, "active"]

docs/examples/usage/usage_data_flow_8.py

@@ -0,0 +1,22 @@
+"""Example 8: Statement Execution."""
+
+__all__ = ("test_statement_execution",)
+
+
+def test_statement_execution() -> None:
+    """Execute a compiled SQL object through SQLSpec."""
+    from sqlspec import SQL, SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
+
+    db_manager = SQLSpec()
+    db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
+
+    # start-example
+    sql_statement = SQL("SELECT ? AS message", "pipeline-complete")
+
+    with db_manager.provide_session(db) as session:
+        result = session.execute(sql_statement)
+        message = result.scalar()
+    # end-example
+
+    assert message == "pipeline-complete"

docs/examples/usage/usage_data_flow_9.py

@@ -0,0 +1,19 @@
+"""Example 9: Driver Execution with session."""
+
+__all__ = ("test_driver_execution",)
+
+
+def test_driver_execution() -> None:
+    """Test driver execution pattern."""
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
+
+    # start-example
+    # Driver receives compiled SQL and parameters
+    db_manager = SQLSpec()
+    db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
+    with db_manager.provide_session(db) as session:
+        result = session.execute("SELECT 'test' as message")
+    # end-example
+
+    assert result is not None