docs: usage drivers and querying examples

.github/workflows/ci.yml

@@ -106,7 +106,7 @@ jobs:
         run: uv sync --all-extras --dev
 
       - name: Test
-        run: uv run pytest -n 2 --dist=loadgroup tests
+        run: uv run pytest -n 2 --dist=loadgroup
 
   # test-windows:
   #   runs-on: windows-latest

docs/examples/quickstart/quickstart_4.py

@@ -25,12 +25,12 @@ class User(BaseModel):
     async with db_manager.provide_session(db) as session:
         await session.execute(
             """
-            CREATE TABLE users (id INTEGER, name TEXT, email TEXT)
+            CREATE TABLE if not exists users (id INTEGER, name TEXT, email TEXT)
             """
         )
-        await session.execute("INSERT INTO users VALUES (?, ?, ?)", 1, "Alice", "[email protected]")
-        user = await session.select_one("SELECT * FROM users WHERE id = ?", 1, schema_type=User)
+        await session.execute("INSERT INTO users VALUES (?, ?, ?)", 100, "Alice", "[email protected]")
+        user = await session.select_one("SELECT * FROM users WHERE id = ?", 100, schema_type=User)
         print(f"User: {user.name}")
     # end-example
 
-    assert user == User(id=1, name="Alice", email="[email protected]")
+    assert user == User(id=100, name="Alice", email="[email protected]")

docs/examples/quickstart/quickstart_6.py

@@ -1,15 +1,21 @@
 __all__ = ("test_quickstart_6",)
 
 
-def test_quickstart_6() -> None:
+from pathlib import Path
+
+
+def test_quickstart_6(tmp_path: Path) -> None:
     # start-example
     from sqlspec import SQLSpec
     from sqlspec.adapters.duckdb import DuckDBConfig
     from sqlspec.adapters.sqlite import SqliteConfig
 
+    app_db = tmp_path / "app.db"
+    analytics_db = tmp_path / "analytics.duckdb"
+
     db_manager = SQLSpec()
-    sqlite_db = db_manager.add_config(SqliteConfig(pool_config={"database": "app.db"}))
-    duckdb_db = db_manager.add_config(DuckDBConfig(pool_config={"database": "analytics.duckdb"}))
+    sqlite_db = db_manager.add_config(SqliteConfig(pool_config={"database": app_db.name}))
+    duckdb_db = db_manager.add_config(DuckDBConfig(pool_config={"database": analytics_db.name}))
 
     with db_manager.provide_session(sqlite_db) as sqlite_session:
         users = sqlite_session.select("SELECT 1")

docs/examples/usage/conftest.py

@@ -1,17 +1,24 @@
 from __future__ import annotations
 
+import os
 from collections.abc import Generator
 
 import pytest
 from pytest_databases.docker.postgres import PostgresService
 
-pytest_plugins = ["pytest_databases.docker.postgres"]
+pytest_plugins = [
+    "pytest_databases.docker.postgres",
+    "pytest_databases.docker.mysql",
+    "pytest_databases.docker.oracle",
+    "pytest_databases.docker.bigquery",
+]
 
 
 @pytest.fixture(scope="session", autouse=True)
 def usage_postgres_env(postgres_service: PostgresService) -> Generator[None, None, None]:
     """Expose Postgres connection settings via env vars for docs examples."""
 
+    os.environ
     patcher = pytest.MonkeyPatch()
     dsn = (
         f"postgresql://{postgres_service.user}:{postgres_service.password}"

docs/examples/usage/usage_configuration_10.py

@@ -1,7 +1,7 @@
 __all__ = ("test_manual_pool",)
 
 
-def test_manual_pool() -> None:
+async def test_manual_pool() -> None:
 
     # start-example
     import os
@@ -10,9 +10,10 @@ def test_manual_pool() -> None:
 
     from sqlspec.adapters.asyncpg import AsyncpgConfig
 
-    pool = asyncpg.create_pool(
+    pool = await asyncpg.create_pool(
         dsn=os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db"), min_size=10, max_size=20
     )
     db = AsyncpgConfig(pool_instance=pool)
     # end-example
     assert db.pool_instance is pool
+    await pool.close()

docs/examples/usage/usage_drivers_and_querying_1.py

@@ -0,0 +1,31 @@
+# Test module converted from docs example - code-block 1
+"""Minimal smoke test for drivers_and_querying example 1."""
+
+import os
+
+from pytest_databases.docker.postgres import PostgresService
+
+__all__ = ("test_importable_1",)
+
+
+async def test_importable_1(postgres_service: PostgresService) -> None:
+    # start-example
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.asyncpg import AsyncpgConfig, AsyncpgPoolConfig
+
+    # Typical driver usage
+    spec = SQLSpec()
+    host = os.environ.get("SQLSPEC_USAGE_PG_HOST", "localhost")
+    port = int(os.environ.get("SQLSPEC_USAGE_PG_PORT", "5432"))
+    user = os.environ.get("SQLSPEC_USAGE_PG_USER", "postgres")
+    password = os.environ.get("SQLSPEC_USAGE_PG_PASSWORD", "postgres")
+    database = os.environ.get("SQLSPEC_USAGE_PG_DATABASE", "sqlspec")
+
+    db = spec.add_config(
+        AsyncpgConfig(
+            pool_config=AsyncpgPoolConfig(host=host, port=port, user=user, password=password, database=database)
+        )
+    )  # Config layer, registers pool
+    async with spec.provide_session(db) as session:  # Session layer
+        await session.execute("SELECT 1")  # Driver layer
+    # end-example

docs/examples/usage/usage_drivers_and_querying_10.py

@@ -0,0 +1,36 @@
+# Test module converted from docs example - code-block 10
+"""Minimal smoke test for drivers_and_querying example 10."""
+
+from pathlib import Path
+
+__all__ = ("test_example_10_duckdb_config",)
+
+
+def test_example_10_duckdb_config() -> None:
+    # start-example
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.duckdb import DuckDBConfig
+
+    spec = SQLSpec()
+    # In-memory
+    config = DuckDBConfig()
+
+    # Persistent
+    config = DuckDBConfig(pool_config={"database": "analytics.duckdb"})
+
+    with spec.provide_session(config) as session:
+        # Create table from Parquet
+        session.execute(f"""
+           CREATE TABLE if not exists users AS
+           SELECT * FROM read_parquet('{Path(__file__).parent.parent / "queries/users.parquet"}')
+       """)
+
+        # Analytical query
+        session.execute("""
+           SELECT date_trunc('day', created_at) as day,
+                  count(*) as user_count
+           FROM users
+           GROUP BY day
+           ORDER BY day
+       """)
+    # end-example

docs/examples/usage/usage_drivers_and_querying_11.py

@@ -0,0 +1,37 @@
+# Test module converted from docs example - code-block 11
+"""Minimal smoke test for drivers_and_querying example 11."""
+
+from pytest_databases.docker.oracle import OracleService
+
+__all__ = ("test_example_11_oracledb_config",)
+
+
+def test_example_11_oracledb_config(oracle_service: OracleService) -> None:
+    # start-example
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.oracledb import OracleSyncConfig
+
+    spec = SQLSpec()
+    config = OracleSyncConfig(
+        pool_config={
+            "user": oracle_service.user,
+            "password": oracle_service.password,
+            "host": oracle_service.host,
+            "port": oracle_service.port,
+            "service_name": oracle_service.service_name,
+        }
+    )
+
+    with spec.provide_session(config) as session:
+        create_table_sql = """CREATE TABLE if not exists employees (
+           employee_id NUMBER PRIMARY KEY,
+           first_name VARCHAR2(50),
+           last_name VARCHAR2(50)
+       )"""
+        session.execute(create_table_sql)
+        session.execute("""
+           INSERT INTO employees (employee_id, first_name, last_name) VALUES (100, 'John', 'Doe')
+       """)
+
+        session.execute("SELECT * FROM employees WHERE employee_id = :id", id=100)
+    # end-example

docs/examples/usage/usage_drivers_and_querying_12.py

@@ -0,0 +1,55 @@
+# Test module converted from docs example - code-block 12
+"""Minimal smoke test for drivers_and_querying example 12."""
+
+import pytest
+
+from sqlspec.adapters.bigquery.driver import BigQueryDriver
+
+__all__ = ("test_example_12_bigquery_config",)
+
+
+@pytest.mark.skip(reason="Requires BigQuery emulator setup with more complex configuration.")
+def test_example_12_bigquery_config(bigquery_service: BigQueryDriver) -> None:
+    # start-example
+    import datetime
+
+    from google.api_core.client_options import ClientOptions
+    from google.auth.credentials import AnonymousCredentials
+
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.bigquery.config import BigQueryConfig
+
+    config = BigQueryConfig(
+        connection_config={
+            "project": bigquery_service.project,
+            "dataset_id": bigquery_service.dataset,
+            "client_options": ClientOptions(api_endpoint=f"http://{bigquery_service.host}:{bigquery_service.port}"),
+            "credentials": AnonymousCredentials(),  # type: ignore[no-untyped-call]
+        }
+    )
+    spec = SQLSpec()
+    with spec.provide_session(config) as bigquery_session:
+        bigquery_session.execute("SELECT 1 AS value")
+
+        # Create the test table
+
+        create_table_query = """
+        CREATE or replace TABLE events (
+            timestamp TIMESTAMP,
+            event_type STRING
+        )
+        """
+        bigquery_session.execute_script(create_table_query)
+
+        print("Executing test query...")
+        bigquery_session.execute(
+            """
+           SELECT DATE(timestamp) as date,
+                  COUNT(*) as events
+           FROM events
+           WHERE timestamp >= @start_date
+           GROUP BY date
+       """,
+            start_date=datetime.date(2025, 1, 1),
+        )
+    # end-example

docs/examples/usage/usage_drivers_and_querying_13.py

@@ -0,0 +1,32 @@
+# Test module converted from docs example - code-block 13
+"""Minimal smoke test for drivers_and_querying example 13."""
+
+__all__ = ("test_example_13_placeholder",)
+
+
+def test_example_13_placeholder() -> None:
+    # start-example
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
+
+    spec = SQLSpec()
+    config = SqliteConfig(pool_config={"database": ":memory:", "timeout": 5.0, "check_same_thread": False})
+    with spec.provide_session(config) as session:
+        create_table_query = (
+            """create table if not exists users (id default int primary key, name varchar(128), email text)"""
+        )
+
+        _ = session.execute(create_table_query)
+        # Examples are documentation snippets; ensure module importable
+        result = session.execute("SELECT * FROM users WHERE id = ?", 1)
+
+        # INSERT query
+        result = session.execute("INSERT INTO users (name, email) VALUES (?, ?)", "Alice", "[email protected]")
+
+        # UPDATE query
+        result = session.execute("UPDATE users SET email = ? WHERE id = ?", "[email protected]", 1)
+        print(f"Updated {result.rows_affected} rows")
+
+        # DELETE query
+        result = session.execute("DELETE FROM users WHERE id = ?", 1)
+    # end-example

docs/examples/usage/usage_drivers_and_querying_14.py

@@ -0,0 +1,97 @@
+"for drivers_and_querying example 14."
+
+import pytest
+
+__all__ = ("test_example_14_placeholder",)
+
+
+def test_example_14_placeholder() -> None:
+    # start-example-1
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
+
+    spec = SQLSpec()
+    config = SqliteConfig(pool_config={"database": ":memory:", "timeout": 5.0, "check_same_thread": False})
+    with spec.provide_session(config) as session:
+        create_table_query = """create table if not exists users (id default int primary key, name varchar(128), email text, status varchar(32))"""
+
+        _ = session.execute(create_table_query)
+        # Batch examples are documentation-only
+
+        # Batch insert
+        session.execute_many(
+            "INSERT INTO users (id, name, email, status) VALUES (?, ?, ?, ?)",
+            [
+                (1, "Alice", "[email protected]", "active"),
+                (2, "Bob", "[email protected]", "inactive"),
+                (3, "Charlie", "[email protected]", "active"),
+            ],
+        )
+        # Batch update
+        session.execute_many("UPDATE users SET status = ? WHERE id = ?", [("inactive", 1), ("active", 2)])
+        # end-example-1
+        # start-example-2
+        results = session.select("SELECT * FROM users")
+        print(results)
+        # Returns list of dictionaries: [{"id": 1, "name": "Alice", ...}, ...]
+        # end-example-2
+        # start-example-3
+        user = session.select_one("SELECT * FROM users WHERE id = ?", 1)
+        print(user)
+        # Returns single dictionary: {"id": 1, "name": "Alice", ...}
+        # Raises NotFoundError if no results
+        # Raises MultipleResultsFoundError if multiple results
+        # end-example-3
+        # start-example-4
+        user = session.select_one_or_none("SELECT * FROM users WHERE email = ?", "[email protected]")
+        # Returns dictionary or None
+        # Raises MultipleResultsFoundError if multiple results
+        # end-example-4
+        # start-example-5
+        session.select_value("SELECT COUNT(*) FROM users")
+        # Returns: 3
+        session.select_value("SELECT MAX(id) FROM users")
+        # Returns: 3
+        # end-example-5
+        # start-example-6
+        result = session.execute("SELECT id, name, email FROM users")
+        # Access raw data
+        result.data  # List of dictionaries
+        result.column_names  # ["id", "name", "email"]
+        result.rows_affected  # For INSERT/UPDATE/DELETE
+        result.operation_type  # "SELECT", "INSERT", etc.
+        # Convenience methods
+        with pytest.raises(ValueError):  # noqa: PT011
+            user = result.one()  # Single row (raises if not exactly 1)
+        with pytest.raises(ValueError):  # noqa: PT011
+            user = result.one_or_none()  # Single row or None
+        with pytest.raises(ValueError):  # noqa: PT011
+            result.scalar()  # First column of first row
+        # end-example-6
+        # start-example-7
+        result = session.execute("SELECT * FROM users")
+        # Get all rows and iterate
+        users = result.all()
+        for user in users:
+            print(f"{user['name']}: {user['email']}")
+        # List comprehension
+        [user["name"] for user in result.all()]
+        # end-example-7
+        # start-example-8
+        from pydantic import BaseModel
+
+        class User(BaseModel):
+            id: int
+            name: str
+            email: str
+
+        # Execute query
+        result = session.execute("SELECT id, name, email FROM users")
+
+        # Map results to typed User instances
+        users: list[User] = result.all(schema_type=User)
+
+        # Or get single typed result
+        user_result = session.execute("SELECT id, name, email FROM users WHERE id = ?", 1)
+        user: User = user_result.one(schema_type=User)
+        # end-example-8