tmp

Author: euri10

docs/examples/usage/conftest.py

@@ -1,59 +1,6 @@
-from collections.abc import Generator
-from typing import Any
-
-import pytest
-from google.api_core.client_options import ClientOptions
-from google.auth.credentials import AnonymousCredentials
-from pytest_databases.docker.bigquery import BigQueryService
-
-from sqlspec.adapters.bigquery.config import BigQueryConfig
-from sqlspec.adapters.bigquery.driver import BigQueryDriver
-
 pytest_plugins = [
     "pytest_databases.docker.postgres",
     "pytest_databases.docker.mysql",
     "pytest_databases.docker.oracle",
     "pytest_databases.docker.bigquery",
 ]
-
-
-@pytest.fixture
-def table_schema_prefix(bigquery_service: "BigQueryService") -> str:
-    """Create a table schema prefix."""
-    return f"`{bigquery_service.project}`.`{bigquery_service.dataset}`"
-
-
-@pytest.fixture
-def bigquery_config(bigquery_service: "BigQueryService", table_schema_prefix: str) -> BigQueryConfig:
-    """Create a BigQuery config object."""
-    return BigQueryConfig(
-        connection_config={
-            "project": bigquery_service.project,
-            "dataset_id": table_schema_prefix,
-            "client_options": ClientOptions(api_endpoint=f"http://{bigquery_service.host}:{bigquery_service.port}"),
-            "credentials": AnonymousCredentials(),  # type: ignore[no-untyped-call]
-        }
-    )
-
-
-@pytest.fixture
-def bigquery_session(bigquery_config: BigQueryConfig) -> Generator[BigQueryDriver, Any, None]:
-    """Create a BigQuery sync session."""
-
-    with bigquery_config.provide_session() as session:
-        yield session
-
-@pytest.fixture
-async def postgres_session_async(postgres_service: "PostgresService") -> Generator[Any, Any, None]:
-    """Create a Postgres async session."""
-    from sqlspec.adapters.asyncpg import AsyncpgConfig
-
-    config = AsyncpgConfig(
-        pool_config={
-            "dsn": f"postgresql://{postgres_service.user}:{postgres_service.password}@{postgres_service.host}:{postgres_service.port}/{postgres_service.database}",
-            "min_size": 1,
-            "max_size": 5,
-        }
-    )
-    async with config.provide_session() as session:
-        yield session

docs/examples/usage/test_drivers_and_querying_12.py

@@ -1,35 +1,48 @@
 # Test module converted from docs example - code-block 12
 """Minimal smoke test for drivers_and_querying example 12."""
 
-from sqlspec.adapters.bigquery import BigQueryDriver
+from sqlspec.adapters.bigquery.driver import BigQueryDriver
 
 
-def test_example_12_bigquery_config(bigquery_session: "BigQueryDriver") -> None:
+def test_example_12_bigquery_config(bigquery_service: BigQueryDriver) -> None:
     import datetime
 
-    from sqlspec import SQLSpec
-
-    SQLSpec()
-
-    bigquery_session.execute("SELECT 1 AS value")
-
-    # Create the test table
+    from google.api_core.client_options import ClientOptions
+    from google.auth.credentials import AnonymousCredentials
 
-    create_table_query = """
-    CREATE TABLE if not exists events (
-        timestamp TIMESTAMP,
-        event_type STRING
+    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]
+        }
     )
-    """
-    bigquery_session.execute_script(create_table_query)
+    spec = SQLSpec()
+    with spec.provide_session(config) as bigquery_session:
+        bigquery_session.execute("SELECT 1 AS value")
+
+        # Create the test table
 
-    bigquery_session.execute(
+        create_table_query = """
+        CREATE or replace TABLE events (
+            timestamp TIMESTAMP,
+            event_type STRING
+        )
         """
-       SELECT DATE(timestamp) as date,
-              COUNT(*) as events
-       FROM events
-       WHERE timestamp >= @start_date
-       GROUP BY date
-   """,
-        start_date=datetime.date(2025, 1, 1),
-    )
+        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),
+        )

docs/examples/usage/test_drivers_and_querying_13.py

@@ -1,7 +1,29 @@
 # Test module converted from docs example - code-block 13
 """Minimal smoke test for drivers_and_querying example 13."""
 
+from sqlspec import SQLSpec
+from sqlspec.adapters.sqlite import SqliteConfig
+
 
 def test_example_13_placeholder() -> None:
-    # Examples are documentation snippets; ensure module importable
-    assert True
+    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", "alice@example.com")
+
+        # UPDATE query
+        result = session.execute("UPDATE users SET email = ? WHERE id = ?", "newemail@example.com", 1)
+        print(f"Updated {result.rows_affected} rows")
+
+        # DELETE query
+        result = session.execute("DELETE FROM users WHERE id = ?", 1)
+        config.close_pool()