feat: oracle update

Author: cofin

docs/examples/quickstart/quickstart_1.py

@@ -1,17 +1,17 @@
-from sqlspec import SQLSpec
-from sqlspec.adapters.sqlite import SqliteConfig
-
 __all__ = ("test_quickstart_1",)
 
 
 def test_quickstart_1() -> None:
-    # Create SQLSpec instance and configure database
+    # start-example
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
+
     db_manager = SQLSpec()
     db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
 
-    # Execute a query
     with db_manager.provide_session(db) as session:
         result = session.execute("SELECT 'Hello, SQLSpec!' as message")
-        print(result.get_first())  # {'message': 'Hello, SQLSpec!'}
+        print(result.get_first())
+    # end-example
 
     assert result.get_first() == {"message": "Hello, SQLSpec!"}

docs/examples/quickstart/quickstart_2.py

@@ -1,44 +1,42 @@
-from sqlspec import SQLSpec
-from sqlspec.adapters.sqlite import SqliteConfig
-
 __all__ = ("test_quickstart_2",)
 
 
 def test_quickstart_2() -> None:
+    # start-example
+    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 table
-        _ = session.execute("""
+        _ = session.execute(
+            """
             CREATE TABLE users (
                 id INTEGER PRIMARY KEY,
                 name TEXT NOT NULL,
                 email TEXT UNIQUE NOT NULL
             )
-        """)
+            """
+        )
 
-        # Insert data
         _ = session.execute("INSERT INTO users (name, email) VALUES (?, ?)", "Alice", "alice@example.com")
 
-        # Insert multiple rows
         _ = session.execute_many(
             "INSERT INTO users (name, email) VALUES (?, ?)",
             [("Bob", "bob@example.com"), ("Charlie", "charlie@example.com")],
         )
 
-        # Query all users
         users = session.select("SELECT * FROM users")
         print(f"All users: {users}")
 
-        # Query single user
-        alice = session.select_one("SELECT * FROM users WHERE name = ?", "Alice")
+        alice = session.select_one_or_none("SELECT * FROM users WHERE name = ?", "Alice")
         print(f"Alice: {alice}")
 
-        # Query scalar value
         count = session.select_value("SELECT COUNT(*) FROM users")
         print(f"Total users: {count}")
+    # end-example
 
-    assert len(users) == 3  # noqa: PLR2004
+    assert len(users) == 3
     assert alice == {"id": 1, "name": "Alice", "email": "alice@example.com"}
-    assert count == 3  # noqa: PLR2004
+    assert count == 3

docs/examples/quickstart/quickstart_3.py

@@ -1,38 +1,36 @@
-from pydantic import BaseModel
+__all__ = ("test_quickstart_3",)
 
-from sqlspec import SQLSpec
-from sqlspec.adapters.sqlite import SqliteConfig
-
-__all__ = ("User",)
 
+def test_quickstart_3() -> None:
+    # start-example
+    from pydantic import BaseModel
 
-class User(BaseModel):
-    id: int
-    name: str
-    email: str
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
 
+    class User(BaseModel):
+        id: int
+        name: str
+        email: str
 
-def test_quickstart_3() -> None:
     db_manager = SQLSpec()
     db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
 
     with db_manager.provide_session(db) as session:
-        # Setup
-        _ = session.execute("""
+        _ = session.execute(
+            """
             CREATE TABLE users (id INTEGER, name TEXT, email TEXT)
-        """)
+            """
+        )
         _ = session.execute("INSERT INTO users VALUES (?, ?, ?)", 1, "Alice", "alice@example.com")
 
-        # Type-safe query - returns User instance
         user = session.select_one("SELECT * FROM users WHERE id = ?", 1, schema_type=User)
+        print(f"User: {user.name} ({user.email})")
 
-        # Now you have type hints and autocomplete!
-        print(f"User: {user.name} ({user.email})")  # IDE knows these fields exist
-
-        # Multiple results
         all_users = session.select("SELECT * FROM users", schema_type=User)
-        for u in all_users:
-            print(f"User: {u.name}")  # Each item is a typed User
+        for typed_user in all_users:
+            print(f"User: {typed_user.name}")
+    # end-example
 
     assert user == User(id=1, name="Alice", email="alice@example.com")
     assert len(all_users) == 1

docs/examples/quickstart/quickstart_4.py

@@ -1,36 +1,36 @@
-from pydantic import BaseModel
+"""Async quickstart example."""
 
-from sqlspec import SQLSpec
-from sqlspec.adapters.aiosqlite import AiosqliteConfig
+import pytest
 
-__all__ = ("User", "test_quickstart_4")
+__all__ = ("test_quickstart_4",)
 
 
-class User(BaseModel):
-    id: int
-    name: str
-    email: str
+@pytest.mark.asyncio
+async def test_quickstart_4() -> None:
+    """Demonstrate async SQLSpec usage."""
+    # start-example
+    from pydantic import BaseModel
 
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.aiosqlite import AiosqliteConfig
+
+    class User(BaseModel):
+        id: int
+        name: str
+        email: str
 
-async def test_quickstart_4() -> None:
     db_manager = SQLSpec()
     db = db_manager.add_config(AiosqliteConfig(pool_config={"database": ":memory:"}))
 
     async with db_manager.provide_session(db) as session:
-        # Create table
-        _ = await session.execute("""
+        await session.execute(
+            """
             CREATE TABLE users (id INTEGER, name TEXT, email TEXT)
-        """)
-
-        # Insert data
-        _ = await session.execute("INSERT INTO users VALUES (?, ?, ?)", 1, "Alice", "alice@example.com")
-
-        # Type-safe async query
+            """
+        )
+        await session.execute("INSERT INTO users VALUES (?, ?, ?)", 1, "Alice", "alice@example.com")
         user = await session.select_one("SELECT * FROM users WHERE id = ?", 1, schema_type=User)
-
         print(f"User: {user.name}")
+    # end-example
 
     assert user == User(id=1, name="Alice", email="alice@example.com")
-    assert isinstance(user, User)
-    assert user.name == "Alice"
-    assert user.email == "alice@example.com"

docs/examples/quickstart/quickstart_5.py

@@ -1,53 +1,58 @@
-import os
-from typing import Any
+"""Async PostgreSQL quickstart example."""
 
-from pydantic import BaseModel
+import pytest
 
-from sqlspec import SQLSpec
-from sqlspec.adapters.asyncpg import AsyncpgConfig
+__all__ = ("test_quickstart_5",)
 
-__all__ = ("User", "test_quickstart_5")
 
-
-class User(BaseModel):
-    id: int
-    name: str
-    email: str
-
-
-def _pool_config() -> "dict[str, Any]":
-    return {
-        "host": os.getenv("SQLSPEC_QUICKSTART_PG_HOST", "localhost"),
-        "port": int(os.getenv("SQLSPEC_QUICKSTART_PG_PORT", "5432")),
-        "user": os.getenv("SQLSPEC_QUICKSTART_PG_USER", "postgres"),
-        "password": os.getenv("SQLSPEC_QUICKSTART_PG_PASSWORD", "postgres"),
-        "database": os.getenv("SQLSPEC_QUICKSTART_PG_DATABASE", "mydb"),
-    }
-
-
-async def _seed_users(session: Any) -> None:
-    await session.execute(
-        """
-        CREATE TABLE IF NOT EXISTS users (
-            id INTEGER PRIMARY KEY,
-            name TEXT NOT NULL,
-            email TEXT NOT NULL
+@pytest.mark.asyncio
+async def test_quickstart_5() -> None:
+    """Demonstrate async PostgreSQL usage with SQLSpec."""
+    # start-example
+    import os
+    from typing import Any
+
+    from pydantic import BaseModel
+
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+    class User(BaseModel):
+        id: int
+        name: str
+        email: str
+
+    def pool_config() -> "dict[str, Any]":
+        return {
+            "host": os.getenv("SQLSPEC_QUICKSTART_PG_HOST", "localhost"),
+            "port": int(os.getenv("SQLSPEC_QUICKSTART_PG_PORT", "5432")),
+            "user": os.getenv("SQLSPEC_QUICKSTART_PG_USER", "postgres"),
+            "password": os.getenv("SQLSPEC_QUICKSTART_PG_PASSWORD", "postgres"),
+            "database": os.getenv("SQLSPEC_QUICKSTART_PG_DATABASE", "mydb"),
+        }
+
+    async def seed_users(session: Any) -> None:
+        await session.execute(
+            """
+            CREATE TABLE IF NOT EXISTS users (
+                id INTEGER PRIMARY KEY,
+                name TEXT NOT NULL,
+                email TEXT NOT NULL
+            )
+            """
+        )
+        await session.execute("TRUNCATE TABLE users")
+        await session.execute(
+            "INSERT INTO users (id, name, email) VALUES ($1, $2, $3)", 1, "Alice", "alice@example.com"
         )
-        """
-    )
-    await session.execute("TRUNCATE TABLE users")
-    await session.execute("INSERT INTO users (id, name, email) VALUES ($1, $2, $3)", 1, "Alice", "alice@example.com")
-
 
-async def test_quickstart_5() -> None:
     db_manager = SQLSpec()
-    db = db_manager.add_config(AsyncpgConfig(pool_config=_pool_config()))
+    db = db_manager.add_config(AsyncpgConfig(pool_config=pool_config()))
 
     async with db_manager.provide_session(db) as session:
-        await _seed_users(session)
-
-        # PostgreSQL uses $1, $2 for parameters instead of ?
+        await seed_users(session)
         user = await session.select_one("SELECT * FROM users WHERE id = $1", 1, schema_type=User)
         print(f"User: {user.name}")
+    # end-example
 
     assert user == User(id=1, name="Alice", email="alice@example.com")

docs/examples/quickstart/quickstart_6.py

@@ -1,23 +1,22 @@
-from sqlspec import SQLSpec
-from sqlspec.adapters.duckdb import DuckDBConfig
-from sqlspec.adapters.sqlite import SqliteConfig
-
 __all__ = ("test_quickstart_6",)
 
 
 def test_quickstart_6() -> None:
-    db_manager = SQLSpec()
+    # start-example
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.duckdb import DuckDBConfig
+    from sqlspec.adapters.sqlite import SqliteConfig
 
-    # Register multiple databases
+    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"}))
 
-    # Use different databases
     with db_manager.provide_session(sqlite_db) as sqlite_session:
         users = sqlite_session.select("SELECT 1")
 
     with db_manager.provide_session(duckdb_db) as duckdb_session:
         analytics = duckdb_session.select("SELECT 1")
+    # end-example
 
     assert isinstance(users, list)
     assert isinstance(analytics, list)

docs/examples/quickstart/quickstart_7.py

@@ -1,43 +1,47 @@
-from sqlspec import SQLSpec
-from sqlspec.adapters.sqlite import SqliteConfig
-
 __all__ = ("test_quickstart_7",)
 
 
 def test_quickstart_7() -> None:
-    db_manager = SQLSpec()
+    # start-example
+    from sqlspec import SQLSpec
+    from sqlspec.adapters.sqlite import SqliteConfig
 
+    db_manager = SQLSpec()
     db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
 
-    # Transaction committed on successful exit
     with db_manager.provide_session(db) as session:
         session.begin()
-        _ = session.execute("""
+        session.execute(
+            """
             CREATE TABLE IF NOT EXISTS users (
                 id INTEGER PRIMARY KEY,
                 name TEXT NOT NULL
             )
-        """)
-        _ = session.execute("""
+            """
+        )
+        session.execute(
+            """
             CREATE TABLE IF NOT EXISTS orders (
                 id INTEGER PRIMARY KEY,
-                    user_name TEXT NOT NULL
-                )
-            """)
-        _ = session.execute("DELETE FROM users")
-        _ = session.execute("DELETE FROM orders")
-        _ = session.execute("INSERT INTO users (name) VALUES (?)", "Alice")
-        _ = session.execute("INSERT INTO orders (user_name) VALUES (?)", "Alice")
+                user_name TEXT NOT NULL
+            )
+            """
+        )
+        session.execute("DELETE FROM users")
+        session.execute("DELETE FROM orders")
+        session.execute("INSERT INTO users (name) VALUES (?)", "Alice")
+        session.execute("INSERT INTO orders (user_name) VALUES (?)", "Alice")
         session.commit()
 
     with db_manager.provide_session(db) as session:
         session.begin()
-        _ = session.execute("INSERT INTO users (name) VALUES (?)", "Bob")
+        session.execute("INSERT INTO users (name) VALUES (?)", "Bob")
         session.rollback()
 
     with db_manager.provide_session(db) as session:
         alice = session.select_one_or_none("SELECT * FROM users WHERE name = ?", "Alice")
         bob = session.select_one_or_none("SELECT * FROM users WHERE name = ?", "Bob")
+    # end-example
 
     assert alice is not None
     assert alice["name"] == "Alice"