docs: usage framework integrations

docs/examples/usage/usage_framework_integrations_1.py

@@ -0,0 +1,26 @@
+__all__ = ("index", "test_app_exists")
+# start-example
+from litestar import Litestar, get
+
+from sqlspec import SQLSpec
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+from sqlspec.extensions.litestar import SQLSpecPlugin
+
+# Configure database and create plugin
+spec = SQLSpec()
+db = spec.add_config(AsyncpgConfig(pool_config={"dsn": "postgresql://localhost/mydb", "min_size": 10, "max_size": 20}))
+sqlspec_plugin = SQLSpecPlugin(sqlspec=spec)
+
+
+@get("/")
+async def index() -> str:
+    return "integrated"
+
+
+# Create Litestar app
+app = Litestar(route_handlers=[index], plugins=[sqlspec_plugin])
+# end-example
+
+
+def test_app_exists() -> None:
+    assert app is not None

docs/examples/usage/usage_framework_integrations_10.py

@@ -0,0 +1,32 @@
+# start-example
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from sqlspec import SQLSpec
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+__all__ = ("lifespan", "test_stub")
+
+
+# Configure database
+spec = SQLSpec()
+db = spec.add_config(AsyncpgConfig(pool_config={"dsn": "postgresql://localhost/mydb", "min_size": 10, "max_size": 20}))
+
+
+# Lifespan context manager
+@asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
+    # Startup
+    yield
+    # Shutdown
+    await spec.close_all_pools()
+
+
+app = FastAPI(lifespan=lifespan)
+# end-example
+
+
+def test_stub() -> None:
+    assert app is not None

docs/examples/usage/usage_framework_integrations_11.py

@@ -0,0 +1,47 @@
+from collections.abc import AsyncGenerator
+from typing import Annotated
+
+from fastapi import Depends, FastAPI
+from litestar.testing import AsyncTestClient
+from pytest_databases.docker.postgres import PostgresService
+
+from sqlspec import SQLSpec
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+from sqlspec.driver import AsyncDriverAdapterBase
+
+__all__ = ("get_db_session", "get_user", "test_stub")
+
+
+async def test_stub(postgres_service: PostgresService) -> None:
+    # start-example
+    app = FastAPI()
+
+    async def get_db_session() -> AsyncGenerator[AsyncDriverAdapterBase, None]:
+        async with spec.provide_session(config) as session:
+            yield session
+
+    # Use in route handlers
+    @app.get("/users/{user_id}")
+    async def get_user(user_id: int, db: Annotated[AsyncDriverAdapterBase, Depends(get_db_session)]) -> dict:
+        result = await db.execute("SELECT id, name, email FROM users_fi11 WHERE id = $1", user_id)
+        return result.one()
+
+    # end-example
+    spec = SQLSpec()
+    config = AsyncpgConfig(
+        pool_config={
+            "host": postgres_service.host,
+            "port": postgres_service.port,
+            "user": postgres_service.user,
+            "password": postgres_service.password,
+            "database": postgres_service.database,
+        }
+    )
+    spec.add_config(config)
+    async with spec.provide_session(config) as session:
+        await session.execute("""CREATE TABLE users_fi11(id integer primary key, name text, email text)""")
+        await session.execute("""INSERT INTO users_fi11(id,name, email) VALUES (1,'toto','[email protected]')""")
+
+    async with AsyncTestClient(app) as client:
+        response = await client.get("/users/1")
+        assert response.json() == {"id": 1, "name": "toto", "email": "[email protected]"}

docs/examples/usage/usage_framework_integrations_12.py

@@ -0,0 +1,54 @@
+# start-example
+from litestar.testing import AsyncTestClient
+
+__all__ = ("create_user", "test_stub")
+
+from collections.abc import AsyncGenerator
+from typing import Annotated, Any
+
+from fastapi import Depends, FastAPI
+from pytest_databases.docker.postgres import PostgresService
+
+from sqlspec import AsyncDriverAdapterBase, SQLSpec
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+
+async def test_stub(postgres_service: PostgresService) -> None:
+    app = FastAPI()
+
+    spec = SQLSpec()
+
+    config = AsyncpgConfig(
+        pool_config={
+            "host": postgres_service.host,
+            "port": postgres_service.port,
+            "user": postgres_service.user,
+            "password": postgres_service.password,
+            "database": postgres_service.database,
+        }
+    )
+
+    async def get_db_session() -> AsyncGenerator[AsyncDriverAdapterBase, None]:
+        async with spec.provide_session(config) as session:
+            yield session
+
+    @app.post("/users")
+    async def create_user(
+        user_data: dict[str, str], db: Annotated[AsyncDriverAdapterBase, Depends(get_db_session)]
+    ) -> dict[str, Any]:
+        async with db.begin_transaction():
+            result = await db.execute(
+                "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id", user_data["name"], user_data["email"]
+            )
+
+            user_id = result.scalar()
+
+            # Additional operations in same transaction
+            await db.execute("INSERT INTO audit_log (action, user_id) VALUES ($1, $2)", "user_created", user_id)
+
+            return result.one()
+
+    # end-example
+
+    async with AsyncTestClient(app) as client:
+        await client.post("/users", json={"name": "bernard", "email": "[email protected]"})

docs/examples/usage/usage_framework_integrations_13.py

@@ -0,0 +1,49 @@
+# start-example
+__all__ = ("generate_report", "get_analytics_db", "get_main_db", "test_stub")
+
+
+# Main database
+from typing import Annotated
+
+from fastapi import Depends, FastAPI
+
+from sqlspec import AsyncDriverAdapterBase, SQLSpec
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+spec = SQLSpec()
+main_db = spec.add_config(AsyncpgConfig(pool_config={"dsn": "postgresql://localhost/main"}))
+
+# Analytics database
+analytics_db = spec.add_config(AsyncpgConfig(pool_config={"dsn": "postgresql://localhost/analytics"}))
+
+
+# Dependency functions
+async def get_main_db():
+    async with spec.provide_session(main_db) as session:
+        yield session
+
+
+async def get_analytics_db():
+    async with spec.provide_session(analytics_db) as session:
+        yield session
+
+
+app = FastAPI()
+
+
+# Use in handlers
+@app.get("/report")
+async def generate_report(
+    main_db: Annotated[AsyncDriverAdapterBase, Depends(get_main_db)],
+    analytics_db: Annotated[AsyncDriverAdapterBase, Depends(get_analytics_db)],
+) -> dict:
+    users = await main_db.execute("SELECT COUNT(*) FROM users")
+    events = await analytics_db.execute("SELECT COUNT(*) FROM events")
+    return {"users": users.scalar(), "events": events.scalar()}
+
+
+# end-example
+
+
+def test_stub() -> None:
+    assert True

docs/examples/usage/usage_framework_integrations_14.py

@@ -0,0 +1,31 @@
+# start-example
+from sanic import Sanic
+
+from sqlspec import SQLSpec
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+__all__ = ("close_db", "test_stub")
+
+
+app = Sanic("MyApp")
+
+# Initialize SQLSpec
+spec = SQLSpec()
+db = spec.add_config(AsyncpgConfig(pool_config={"dsn": "postgresql://localhost/db"}))
+
+# Store in app context
+app.ctx.sqlspec = spec
+app.ctx.db_config = db
+
+
+# Cleanup on shutdown
+@app.before_server_stop
+async def close_db(app, loop) -> None:
+    await app.ctx.sqlspec.close_all_pools()
+
+
+# end-example
+
+
+def test_stub() -> None:
+    assert app is not None

docs/examples/usage/usage_framework_integrations_15.py

@@ -0,0 +1,23 @@
+# start-example
+__all__ = ("get_user", "test_stub")
+
+
+import json
+
+from fastapi import FastAPI, Request
+
+app = FastAPI()
+
+
+@app.get("/users/<user_id:int>")
+async def get_user(request: Request, user_id: int):
+    async with request.app.ctx.sqlspec.provide_session(request.app.ctx.db_config) as db:
+        result = await db.execute("SELECT id, name, email FROM users WHERE id = $1", user_id)
+        return json(result.one())
+
+
+# end-example
+
+
+def test_stub() -> None:
+    assert True

docs/examples/usage/usage_framework_integrations_16.py

@@ -0,0 +1,32 @@
+__all__ = ("add_db_session", "cleanup_db_session", "list_users", "test_stub")
+# start-example
+import json
+
+from fastapi import FastAPI, Request
+
+app = FastAPI()
+
+
+@app.middleware("request")
+async def add_db_session(request) -> None:
+    request.ctx.db = await request.app.ctx.sqlspec.provide_session(request.app.ctx.db_config).__aenter__()
+
+
+@app.middleware("response")
+async def cleanup_db_session(request, response) -> None:
+    if hasattr(request.ctx, "db"):
+        await request.ctx.db.__aexit__(None, None, None)
+
+
+# Use in handlers
+@app.get("/users")
+async def list_users(request: Request):
+    result = await request.ctx.db.execute("SELECT * FROM users")
+    return json(result.rows)
+
+
+# end-example
+
+
+def test_stub() -> None:
+    assert True

docs/examples/usage/usage_framework_integrations_17.py

@@ -0,0 +1,19 @@
+# start-example
+from flask import Flask
+
+from sqlspec import SQLSpec
+from sqlspec.adapters.sqlite import SqliteConfig
+
+__all__ = ("test_stub",)
+
+
+app = Flask(__name__)
+
+# Initialize SQLSpec
+spec = SQLSpec()
+db = spec.add_config(SqliteConfig(pool_config={"database": "app.db"}))
+# end-example
+
+
+def test_stub() -> None:
+    assert app is not None

docs/examples/usage/usage_framework_integrations_18.py

@@ -0,0 +1,33 @@
+__all__ = ("close_db", "get_db", "get_user", "test_stub")
+# start-example
+from flask import Flask
+
+app = Flask(__name__)
+
+
+def get_db():
+    if "db" not in g:
+        g.db = spec.provide_session(db).__enter__()
+    return g.db
+
+
+@app.teardown_appcontext
+def close_db(error) -> None:
+    db = g.pop("db", None)
+    if db is not None:
+        db.__exit__(None, None, None)
+
+
+# Use in routes
+@app.route("/users/<int:user_id>")
+def get_user(user_id):
+    db = get_db()
+    result = db.execute("SELECT * FROM users WHERE id = ?", user_id)
+    return result.one()
+
+
+# end-example
+
+
+def test_stub() -> None:
+    assert True

docs/examples/usage/usage_framework_integrations_19.py

@@ -0,0 +1,33 @@
+# start-example
+__all__ = ("DatabaseSession", "example_usage", "test_stub")
+
+
+from sqlspec import SQLSpec
+
+
+class DatabaseSession:
+    def __init__(self, spec: SQLSpec, config) -> None:
+        self.spec = spec
+        self.config = config
+        self.session = None
+
+    async def __aenter__(self):
+        self.session = await self.spec.provide_session(self.config).__aenter__()
+        return self.session
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        if self.session:
+            await self.session.__aexit__(exc_type, exc_val, exc_tb)
+
+
+# Usage
+async def example_usage() -> None:
+    async with DatabaseSession(spec, config) as db:
+        await db.execute("SELECT * FROM users")
+
+
+# end-example
+
+
+def test_stub() -> None:
+    assert True