initial fastapi-sqlalchemy

2026-06-17 21:45:54 -05:00
parent 6e32955533
commit 22e5357ffb
4 changed files with 354 additions and 0 deletions
@@ -0,0 +1,221 @@
 ---
 name: fastapi-async-sqlalchemy-modernization
 description: 'Create a step-by-step modernization plan for an existing FastAPI app using SQLAlchemy async patterns, context managers, and AsyncExitStack. Use when: planning migration from legacy DB setup, standardizing async engine/session lifecycles, defining transaction boundaries, and aligning with SQLAlchemy 2.x best practices.'
 argument-hint: 'What is your current FastAPI + SQLAlchemy setup (sync/async driver, session pattern, lifespan usage, and deployment model)?'
 ---
 # FastAPI Async SQLAlchemy Modernization Plan
 Create an implementation-ready plan that brings an existing FastAPI application in line with modern async SQLAlchemy practices, with explicit resource lifecycles and deterministic cleanup using async context managers and AsyncExitStack.
 Primary targets: PostgreSQL with asyncpg and SQLite with aiosqlite.
 ## When to Use
 - Existing FastAPI app has ad hoc database setup or mixed sync/async access.
 - Session management is inconsistent across routes/services.
 - Lifespan startup and shutdown work is spread across globals and side effects.
 - Team needs a migration plan first, not immediate large-scale rewrites.
 ## Outcome
 Produce a practical modernization plan with:
 - Current-state gap assessment.
 - Target architecture for engine/session/transaction lifecycle.
 - Branch-based migration path (low-risk staged rollout).
 - Quality gates and completion checks.
 - Risks, rollback strategy, and test plan.
 ## Top-Level Concepts
 Use these concepts as the planning backbone:
 1. Engine lifecycle and ownership:
   One AsyncEngine per process for each DB URL, created once and disposed explicitly when the app lifecycle ends.
 2. Session factory and scope:
   Use async_sessionmaker for configuration; create one AsyncSession per request or unit-of-work, never shared across concurrent tasks.
 3. Transaction boundaries:
   Prefer context-managed begin blocks for write units and explicit read-only sessions for queries.
 4. Lifespan composition:
   Compose startup/shutdown resources with AsyncExitStack so cleanup is deterministic and ordered.
 5. Dependency injection:
   Provide sessions via FastAPI dependencies with async generators/context managers, not globals.
 6. Implicit I/O control in ORM:
   Avoid accidental lazy loads; use explicit eager-loading/refresh strategies for asyncio safety.
 7. Observability and resilience:
   Add pool/connection settings, logging, timeout, and health checks as first-class plan items.
 ## Decision Points
 Use these branching decisions before proposing migration steps.
 | Decision | Branch A | Branch B |
 |---|---|---|
 | DB driver | Already async driver (e.g. asyncpg, aiosqlite): modernize in place | Sync driver: plan driver migration first |
 | ORM usage | Already ORM 2.x style (`select`, `session.execute`) | Legacy Query API: add compatibility stage and refactor incrementally |
 | Session scope | Request-scoped already | Global/shared sessions found: prioritize session-scope fix first |
 | Lifespan | Existing FastAPI lifespan hook | No lifespan hook: introduce lifespan before broader DB changes |
 | Concurrency | Background jobs/tasks use DB | No background DB use |
 | Transaction style | Explicit context-managed transactions | Implicit/autobegin side effects |
 ## Procedure
 ### Step 0: Audit Current State
 Inventory the app and write a concise gap list.
 - Engine creation location(s) and count.
 - Driver URL(s) and async compatibility.
 - Session creation patterns in routes/services/background tasks.
 - Transaction handling style (explicit begin/commit/rollback vs implicit).
 - Lifespan startup/shutdown and cleanup behavior.
 - ORM loading patterns that may trigger implicit I/O.
 Completion check: every DB touchpoint is mapped to its engine, session, and transaction source.
 ### Step 1: Define the Target Runtime Model
 Define one canonical model to migrate toward.
 - Create AsyncEngine once per process.
 - Configure async_sessionmaker once.
 - Use per-request AsyncSession dependency.
 - Keep one AsyncSession per concurrent task.
 - Use context-managed transactions for writes.
 Completion check: architecture diagram can explain where engine/session are created, used, and closed.
 ### Step 2: Plan Engine Modernization
 Plan engine creation and pool behavior.
 - Use `create_async_engine()` with async dialect URL.
 - Standardize pool settings and pre-ping strategy where relevant.
 - Decide isolation level strategy at engine level (avoid ad hoc per-operation switching unless justified).
 - Define explicit disposal policy for short-lived scopes and tests.
 Completion check: engine configuration is centralized and no per-request engine creation remains.
 ### Step 3: Plan Session Lifecycle Modernization
 Define session factory and request dependency pattern.
 - Build `async_sessionmaker(engine, expire_on_commit=False)` unless a strict reason says otherwise.
 - Provide session via dependency that yields exactly one AsyncSession.
 - Explicitly prohibit sharing a single AsyncSession across concurrent tasks.
 - Prefer direct dependency passing over async_scoped_session for new designs.
 Completion check: all route/service entry points receive a session from one canonical dependency.
 ### Step 4: Plan Transaction Demarcation
 Establish consistent write and read behavior.
 - Writes: `async with session.begin(): ...` for atomic units.
 - Reads: execute in managed session context with explicit loader options.
 - Nested/SAVEPOINT use only where required; call out backend caveats.
 - Define rollback behavior for service-layer exceptions.
 Completion check: every mutating use case has a declared transaction boundary.
 ### Step 5: Compose Lifespan with AsyncExitStack
 Use async context composition as the preferred orchestration pattern.
 ```python
 from contextlib import AsyncExitStack, asynccontextmanager
 from fastapi import FastAPI
@asynccontextmanager
 async def lifespan(app: FastAPI):
    async with AsyncExitStack() as stack:
        # Compose resources in acquisition order; cleanup is automatic in reverse order.
        engine = create_async_engine(settings.database_url)
        stack.push_async_callback(engine.dispose)
        session_factory = async_sessionmaker(engine, expire_on_commit=False)
        app.state.session_factory = session_factory
        # Add other async resources with stack.enter_async_context(...) as needed.
        yield
 ```
 Planning rules:
 - Register every acquired resource with AsyncExitStack at acquisition time.
 - Prefer `enter_async_context()` for resources that already expose async context managers.
 - Prefer `push_async_callback()` for async cleanup callables.
 - Keep resource ownership in lifespan, not in route handlers.
 Completion check: startup/shutdown ordering is explicit and deterministic.
 ### Step 6: Prevent Implicit ORM I/O Under Asyncio (Advisory Mode)
 Plan for explicit loading behavior, but treat this as progressive guidance rather than a hard gate.
 - Recommend eager-loading strategies (for example selectin-style loading) where relationship access is required.
 - For lazy/deferred attributes, define explicit awaitable or refresh paths on high-risk and high-traffic paths first.
 - Document model-level defaults and known exceptions so teams can migrate incrementally.
 Completion check: critical request paths have explicit loading plans; non-critical paths have tracked follow-up items.
 ### Step 7: Testing and Verification Plan
 Create modernization quality gates.
 - Unit tests for session dependency and transaction behavior.
 - Integration tests for commit/rollback semantics.
 - Concurrency tests confirming one-session-per-task behavior.
 - Lifespan tests verifying cleanup calls and ordering.
 - Health/readiness tests including DB connectivity checks.
 Completion check: all quality gates pass under the target async configuration.
 ### Step 8: Rollout Strategy
 Plan low-risk migration phases.
 1. Introduce centralized engine/session factory and lifespan orchestration.
 2. Migrate read paths to new session dependency.
 3. Migrate write paths to explicit transaction blocks.
 4. Remove legacy globals/helpers and dead code.
 5. Enable stricter linting/review checks for forbidden patterns.
 Completion check: no legacy session/engine creation path remains in production code.
 ## Quality Criteria
 A plan is complete only when it includes:
 - Clear current vs target architecture.
 - Branch decisions with rationale.
 - Explicit context-manager patterns for resource ownership.
 - AsyncExitStack composition strategy.
 - Transaction policy and exception behavior.
 - Concrete tests and rollout checkpoints.
 - A documented advisory backlog for non-critical implicit I/O improvements.
 ## Anti-Patterns to Flag
 - Creating engines inside request handlers.
 - Sharing one AsyncSession across concurrent tasks.
 - Implicit commit/rollback behavior with unclear ownership.
 - Global mutable session state.
 - Lifespan cleanup that depends on implicit garbage collection.
 ## Output Contract
 Return the plan as:
 1. Current-state gap summary.
 2. Target architecture summary.
 3. Phased migration checklist with branch notes.
 4. Risk register and rollback approach.
 5. Verification matrix (tests + operational checks).
 ## References
 - SQLAlchemy engine/connections: https://docs.sqlalchemy.org/en/21/core/connections.html
 - SQLAlchemy asyncio extension: https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html
 - Python async context managers and AsyncExitStack: https://docs.python.org/3/library/contextlib.html
@@ -0,0 +1,133 @@
 # Async SQLAlchemy Engine
 Source:
 - https://docs.sqlalchemy.org/en/21/core/connections.html
 - https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html
 - https://docs.sqlalchemy.org/en/21/core/pooling.html#pooling-multiprocessing
 - https://fastapi.tiangolo.com/advanced/events/
 ---
 ## Engine Ownership Model
 Create one async engine per process per database URL and keep it for the app lifetime.
 - SQLAlchemy guidance: the engine is intended as a long-lived, concurrent registry over pooled DB connections, not a per-request object.
 - In FastAPI, app startup and shutdown ownership belongs in lifespan.
 - Use `FastAPI(lifespan=...)` (not startup/shutdown events) for modern lifecycle wiring.
 Practical rule:
 - Exactly one `create_async_engine(...)` call in app bootstrap code.
 - Zero `create_async_engine(...)` calls in request handlers.
 ---
 ## Canonical Lifespan Pattern (AsyncExitStack)
 Use `@asynccontextmanager` + `AsyncExitStack` to make teardown deterministic and composable.
 ```python
 from contextlib import AsyncExitStack, asynccontextmanager
 from fastapi import FastAPI
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
@asynccontextmanager
 async def lifespan(app: FastAPI):
    async with AsyncExitStack() as stack:
        engine: AsyncEngine = create_async_engine(
            app.state.settings.database_url,
            pool_pre_ping=True,
            # Optional examples:
            # echo=app.state.settings.sql_echo,
            # pool_size=10,
            # max_overflow=20,
        )
        app.state.engine = engine
        # Ensure engine disposal always runs at shutdown.
        stack.push_async_callback(engine.dispose)
        yield
 app = FastAPI(lifespan=lifespan)
 ```
 Why this pattern:
 - FastAPI executes code before `yield` at startup and after `yield` at shutdown.
 - `AsyncExitStack` lets you register multiple async cleanups in one place while preserving order.
 - Explicit disposal (directly awaited or via `AsyncExitStack` callback) avoids event-loop-closed warnings when objects fall out of scope.
 ---
 ## Driver URLs (Project Requirement: asyncpg + aiosqlite)
 Use SQLAlchemy async driver URLs:
 - PostgreSQL: `postgresql+asyncpg://user:pass@host:5432/dbname`
 - SQLite: `sqlite+aiosqlite:///./app.db`
 Notes:
 - Do not mix sync drivers (for example `psycopg2`) with `create_async_engine()`.
 - Keep URL construction centralized in settings/config, not in feature modules.
 ---
 ## Pooling Defaults and Tuning
 Default behavior is usually correct first:
 - Async engines use async-compatible pooling (`AsyncAdaptedQueuePool`) by default.
 - Start with defaults, then tune from observed load (`pool_size`, `max_overflow`, `pool_timeout`, `pool_recycle`).
 - Enable `pool_pre_ping=True` for safer stale-connection handling in long-running services.
 When to switch pool strategy:
 - `NullPool` if you explicitly need no pooling (special environments, some tests, or strict cross-loop constraints).
 - Keep in mind this increases connect/disconnect churn.
 ---
 ## Disposal Semantics
 `engine.dispose()` replaces/disposes the pool, but only checked-in connections are immediately closed.
 Rules:
 - Dispose when the app is shutting down.
 - Dispose before reusing an engine across event loops.
 - In forked child-process initialization, use `engine.dispose(close=False)` (sync API guidance) so child processes do not touch parent-held connections.
 Avoid relying on garbage collection for engine cleanup in async code.
 ---
 ## Event Loop and Process Boundaries
 Do not share pooled connections across boundaries:
 - Multiple event loops: do not reuse the same pooled async engine across loops unless you intentionally disable pooling (`NullPool`) or dispose before handoff.
 - Multiprocessing/fork: pooled connections must not be inherited for active use across process boundaries.
 This prevents broken socket state and cross-process connection corruption.
 ---
 ## What Not to Do
 - Create an engine inside every request dependency.
 - Create/dispose engines inside repository methods.
 - Keep engine creation as a hidden side effect of import-time module globals.
 - Use deprecated FastAPI startup/shutdown events together with lifespan.
 ---
 ## Engine Design Checklist
 - One engine per process per DB URL.
 - Engine created in lifespan startup.
 - Engine disposed in lifespan shutdown.
 - Async driver URL matches backend (`asyncpg` or `aiosqlite`).
 - Pooling strategy is explicit for non-default needs.
 - No request-path engine creation.