prompt toc update

Dockerfile

@@ -47,7 +47,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
     --mount=type=bind,source=uv.lock,target=uv.lock \
     --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
     --mount=type=bind,source=src/,target=src/ \
-    uv sync --no-editable
+    uv sync --no-editable --refresh-package prompts
 
 USER appuser
 

docker-compose.yaml

@@ -6,5 +6,3 @@ services:
     restart: unless-stopped
     ports:
       - "8765:8765"
-    volumes:
-      - ./docs:/app/src/personal_mcp/docs

docs/prompts/jsfiddle-page-layout/PROMPT.md

@@ -0,0 +1,66 @@
+---
+name: jsfiddle-page-layout
+description: Create a responsive sample page layout for a user-supplied domain and return paste-ready HTML and CSS for JSFiddle.
+x-personal-mcp:
+  id: jsfiddle-page-layout
+  version: 1.1.0
+  tags:
+    - frontend
+    - html
+    - css
+    - jsfiddle
+    - layout
+    - prototyping
+    - prompts
+  capabilities:
+    - resource://prompts/jsfiddle-page-layout/document
+---
+
+# JSFiddle Page Layout
+
+Create a polished sample page layout for the supplied domain. The result must run by pasting the markup and styles into the [JSFiddle](https://jsfiddle.net/) HTML and CSS panes.
+
+## Inputs
+
+1. `domain`: the product, service, organization, or subject represented by the page, including its intended audience when known
+2. `layout_brief`: optional page type, required sections, content priorities, visual direction, or constraints
+
+## Workflow
+
+1. Infer the page's primary purpose, audience, content hierarchy, and most important user action from the inputs.
+2. If the domain does not provide enough information to choose a useful page type or primary action, ask one concise clarification question before generating code.
+3. Choose a visual direction and information density appropriate to the domain. Build the usable page itself, not a marketing explanation of the page.
+4. Write semantic HTML with realistic domain-specific sample content. Do not use placeholder text such as lorem ipsum.
+5. Build the layout with modern CSS, using [CSS Grid](https://css-tricks.com/complete-guide-css-grid-layout/) for two-dimensional page structure and [Flexbox](https://css-tricks.com/snippets/css/a-guide-to-flexbox/) for one-dimensional alignment where each fits naturally.
+6. Make the page responsive at narrow mobile and desktop widths without horizontal overflow, overlapping content, or clipped text.
+7. Keep the example self-contained. Use no JavaScript, build tools, external stylesheets, images, or icon libraries unless the layout brief explicitly requires them.
+8. Include accessible landmarks, heading order, labels, focus styles, color contrast, and reduced-motion handling when animation is present.
+9. Use CSS custom properties for the color, typography, spacing, border, and shadow system. Avoid generic framework styling and tailor the visual language to the domain.
+
+## Design References
+
+Use these references as comparative guidance, not as templates to copy. Select principles that fit the domain and layout brief, and do not reproduce a vendor's visual language unless the user requests it.
+
+1. [Material Design 3 foundations](https://m3.material.io/foundations) for current approaches to layout, interaction states, design tokens, and adaptable UI systems.
+2. [Apple Human Interface Guidelines](https://developer.apple.com/design/human-interface-guidelines/) for contemporary principles covering hierarchy, typography, controls, and platform-aware interaction.
+3. [web.dev responsive web design basics](https://web.dev/articles/responsive-web-design-basics) for content-led breakpoints, flexible layouts, and input-aware responsiveness.
+4. [Web Content Accessibility Guidelines (WCAG) 2.2](https://www.w3.org/TR/WCAG22/) as the accessibility baseline for structure, contrast, focus, reflow, and target sizing.
+
+## Output Contract
+
+Return exactly two fenced code blocks in this order:
+
+1. An `html` block containing only the content for JSFiddle's HTML pane.
+2. A `css` block containing only the content for JSFiddle's CSS pane.
+
+Do not include setup instructions, design commentary, JavaScript, or prose outside the two code blocks.
+
+## Quality Rules
+
+1. Prefer semantic elements such as `header`, `nav`, `main`, `section`, `article`, `aside`, and `footer` when they match the content.
+2. Reserve large display type for a true hero or primary page title; keep operational interfaces compact and easy to scan.
+3. Use cards only for repeated items or genuinely framed tools. Do not place cards inside cards.
+4. Use stable responsive constraints for grids, controls, media, and navigation so dynamic content does not shift the layout unexpectedly.
+5. Avoid decorative gradients, floating color blobs, excessive rounding, and one-note palettes unless they are explicitly appropriate to the domain.
+6. Ensure controls look and behave like their purpose, with visible hover and keyboard-focus states.
+7. Keep all visible copy relevant to the fictional domain rather than describing the mockup or its implementation.

docs/prompts/nicegui-component-extraction/PROMPT.md

@@ -0,0 +1,95 @@
+---
+name: nicegui-component-extraction
+description: Extract a user-selected component from a JSFiddle page layout and implement it as a reusable NiceGUI render function with responsive styling and typed bindable state where needed.
+x-personal-mcp:
+  id: nicegui-component-extraction
+  version: 1.0.0
+  tags:
+    - nicegui
+    - components
+    - frontend
+    - refactoring
+    - jsfiddle
+    - prompts
+  capabilities:
+    - resource://prompts/nicegui-component-extraction/document
+  arguments:
+    component:
+      title: Component
+      description: Component or page region to extract, identified by its visible label, semantic role, or selector.
+      required: true
+    source_layout:
+      title: Source layout
+      description: Optional HTML and CSS from the JSFiddle page layout prompt; when omitted, use the latest applicable output in the conversation.
+      required: false
+    target_location:
+      title: Target location
+      description: Optional target NiceGUI page, module, or package in which to create and integrate the component.
+      required: false
+    behavior_requirements:
+      title: Behavior requirements
+      description: Optional interactions, state, callbacks, or content variations the extracted component must support.
+      required: false
+---
+
+# NiceGUI Component Extraction
+
+Extract one user-selected component from the output of the [JSFiddle Page Layout](../jsfiddle-page-layout/PROMPT.md) prompt and implement it as a reusable NiceGUI component in the target repository.
+
+## Inputs
+
+1. `component`: required visible label, semantic role, or selector identifying the component to extract
+2. `source_layout`: optional HTML and CSS; when omitted, use the latest applicable JSFiddle page layout output in the conversation
+3. `target_location`: optional target page, module, or package; infer it from the repository when omitted
+4. `behavior_requirements`: optional interactions, state, callbacks, or content variations
+
+If the selected component or source layout cannot be identified unambiguously, ask one concise clarification question before editing.
+
+## Required References
+
+Apply both references before implementation:
+
+1. Component boundaries, responsive layout, Quasar props, Tailwind utilities, and shared CSS: [NiceGUI Page Layout and Styling](../../skills/nicegui/references/architecture-and-styling.md)
+2. Typed UI state, propagation, mutable defaults, binding strictness, and version checks: [Binding Dataclasses Deep Dive](../../skills/nicegui/references/binding-dataclasses.md)
+
+## Workflow
+
+1. Locate the selected region in the source HTML and CSS, including its responsive rules, states, and dependencies on surrounding layout.
+2. Inspect the target repository's NiceGUI version, package structure, component conventions, shared CSS loading, and nearest page call site.
+3. Define the smallest reusable API for the component:
+   - name the public function `render_<component_name>` using snake_case
+   - accept content, typed state, and event callbacks as explicit parameters
+   - keep business rules, persistence, and service access outside the component
+   - preserve an established return-value convention; otherwise return the component's root NiceGUI element
+4. Translate semantic HTML into native NiceGUI and Quasar elements. Do not embed the original page wholesale with `ui.html` when standard components express the structure.
+5. Recreate only the CSS needed by the extracted component:
+   - use Quasar props for component appearance and behavior
+   - use NiceGUI classes and Tailwind utilities for spacing, sizing, alignment, and responsive layout
+   - use scoped shared CSS only where props and utilities are insufficient
+   - do not override Quasar field internals or duplicate globally loaded styles
+6. Model editable or shared component state with a typed `@binding.bindable_dataclass` only when binding improves the interaction:
+   - use `field(default_factory=...)` for mutable defaults
+   - scope state to the appropriate page, client, or user
+   - keep binding transforms pure and inexpensive
+   - assign updated collections back to bound fields instead of relying on in-place mutation
+7. Integrate the render function at the nearest target page or call site without moving unrelated page composition or domain logic into the component.
+8. Preserve accessibility, focus behavior, text wrapping, stable dimensions, and the source layout's visual hierarchy.
+9. Run the narrowest available tests, lint, and type checks for the changed files. For visual components, verify representative mobile, landscape desktop, and portrait desktop viewports when browser tooling is available.
+
+## Output Contract
+
+Complete the implementation in the target repository, then report:
+
+1. Files created or updated.
+2. The `render_*` function signature and its state or callback contract.
+3. Any deliberate visual or interaction differences from the JSFiddle source.
+4. Validation commands and outcomes, including viewport checks when performed.
+
+## Quality Rules
+
+1. Extract exactly the requested component and its necessary local dependencies.
+2. Prefer the target repository's established patterns over introducing a new abstraction style.
+3. Keep the component presentation-focused and reusable across pages with compatible data.
+4. Do not add a bindable dataclass for static content or event-local state that is clearer as ordinary parameters.
+5. Do not create a second component tree for mobile; use responsive classes and stable layout constraints.
+6. Keep custom CSS tokenized, scoped to the component, and loaded once by the application's composition layer.

docs/skills/async-fastapi-sqlmodel/SKILL.md

@@ -1,9 +1,9 @@
 ---
 name: async-fastapi-sqlmodel
-description: 'Explain and apply async database principles for FastAPI, SQLAlchemy 2.x, and SQLModel. Use when: learning or reviewing AsyncEngine and AsyncSession lifecycles, FastAPI lifespan and yield dependencies, transaction boundaries, concurrency safety, implicit ORM I/O, AsyncExitStack, pooling, testing, or SQLModel integration.'
+description: 'Explain and apply async database principles for FastAPI, SQLAlchemy 2.x, and SQLModel. Use when: learning or reviewing cached AsyncEngine and session-factory lifecycles, AsyncSession scopes and injection, FastAPI lifespan and yield dependencies, transaction boundaries, concurrency safety, implicit ORM I/O, pooling, testing, or SQLModel integration.'
 x-personal-mcp:
   id: async-fastapi-sqlmodel
-  version: 1.1.0
+  version: 1.2.0
   tags:
     - fastapi
     - sqlalchemy
@@ -24,6 +24,8 @@ Use this skill to explain how an async database layer works, why the recommended
 
 Primary targets: PostgreSQL with asyncpg and SQLite with aiosqlite.
 
+Engine and session mechanics mirror the [`nicegui-db` template repository](https://forgejo.john-stream.com/john/nicegui-db). Treat that template as the implementation baseline, then explain the rationale, lifecycle constraints, and tradeoffs behind its cached engines, session factories, context managers, dependency wiring, and `with_session` decorator. Source-specific claims in the references link to the reviewed template commit so behavior remains auditable as the template evolves.
+
 ## When to Use
 
 - Explain an async engine, session factory, session, connection, or transaction.
@@ -51,7 +53,7 @@ Keep three ownership scopes distinct:
 
 | Scope | Object | Purpose | Typical owner |
 |---|---|---|---|
-| Application process | `AsyncEngine` and `async_sessionmaker` | Dialect, connection pool, and repeatable session configuration | FastAPI lifespan |
+| Application process | Cached `AsyncEngine` and lifespan-owned `async_sessionmaker` | Dialect, connection pool, schema initialization, and repeatable session configuration | FastAPI lifespan |
 | Request or concurrent task | `AsyncSession` | Mutable ORM identity map and transactional state | A `yield` dependency or explicit unit of work |
 | Atomic operation | `SessionTransaction` | Commit all changes together or roll them back together | Service or use-case boundary |
 
@@ -61,16 +63,16 @@ The engine is a long-lived factory and pool, not a single database connection. T
 
 ### Match lifetime to ownership
 
-- Create one `AsyncEngine` per process and database configuration in the normal case.
+- Resolve one cached `AsyncEngine` per database URL during the active application lifecycle.
-- Dispose it explicitly in an awaitable shutdown path; garbage collection cannot reliably await async driver cleanup.
+- Enter one owning engine scope per URL; initialize registered SQLModel metadata by default, then dispose the engine and clear cached resolution on exit.
-- Configure `async_sessionmaker` once and call it to create short-lived sessions.
+- Configure the application `async_sessionmaker` inside the engine lifecycle; use the template's cached factory resolver only for standalone helpers that cannot receive the application factory.
 - Close each session deterministically with `async with` or a FastAPI dependency that yields once.
 
 See [engine lifecycle](references/engine.md) and [session management](references/session.md).
 
 ### Isolate mutable session state
 
-An `AsyncSession` represents one stateful transaction in progress. Never use one session in multiple concurrent tasks, including branches of `asyncio.gather()`. Give each task its own session and pass sessions explicitly rather than relying on mutable scoped globals.
+An `AsyncSession` represents one stateful transaction in progress. Never use one session in multiple concurrent tasks, including branches of `asyncio.gather()`. Give each task its own session. Template-style `@with_session` functions inject one only when the `session` argument is omitted; a supplied session remains caller-owned, and explicit `None` is forwarded unchanged.
 
 See [session management](references/session.md).
 
@@ -92,9 +94,9 @@ See [transaction boundaries](references/transactions.md).
 
 FastAPI lifespan owns resources shared by many requests. A dependency with one `yield` owns request-scoped resources and runs cleanup after use. These are related context-manager mechanisms but solve different lifetime problems.
 
-Use `AsyncExitStack` when lifespan acquires a variable, conditional, or mixed collection of context-managed resources. It records cleanup as resources are acquired and unwinds callbacks in reverse order. A single engine with one cleanup callback can use a plain `try/finally`; `AsyncExitStack` is a composition tool, not a requirement.
+Use `AsyncExitStack` when lifespan acquires a variable, conditional, or mixed collection of context-managed resources. It records cleanup as resources are acquired and unwinds callbacks in reverse order. A single engine should use the direct engine context manager; `AsyncExitStack` is a composition tool, not a requirement.
 
-See [engine lifecycle](references/engine.md).
+See [FastAPI database integration](references/fastapi.md).
 
 ### Use SQLModel as the primary modeling layer
 
@@ -110,6 +112,14 @@ Pool sizing, overflow, recycle, pre-ping, isolation, statement timeouts, and hea
 
 See [observability and resilience](references/observability.md).
 
+### Test through the production seam
+
+Keep the production engine and session-factory construction path intact in tests. Select a dedicated PostgreSQL, local SQLite, or in-memory SQLite URL at that seam, then override the request-session dependency only for the test lifetime. Use a test-scoped outer transaction with SAVEPOINT-backed session commits when application code calls `commit()`; it exercises normal transaction behavior while cleanup remains deterministic.
+
+In-memory SQLite is suitable for serial tests. For multiple simultaneous sessions, use a named shared-cache SQLite URL or a temporary file, and retain PostgreSQL integration coverage for PostgreSQL-specific behavior.
+
+See [database testing and fixture data](references/testing.md).
+
 ## Reference Map
 
 | Concept | Reference |
@@ -117,41 +127,19 @@ See [observability and resilience](references/observability.md).
 | Engine lifecycle and ownership | [Engine lifecycle reference](references/engine.md) |
 | Session factory and scope | [Session management reference](references/session.md) |
 | Transaction boundaries | [Transaction boundaries reference](references/transactions.md) |
-| Lifespan composition | [Engine lifecycle reference](references/engine.md) |
+| FastAPI lifespan composition | [FastAPI integration reference](references/fastapi.md) |
-| Dependency injection | [Session management reference](references/session.md) |
+| FastAPI dependency injection | [FastAPI integration reference](references/fastapi.md) |
 | Implicit I/O control in ORM | [Implicit I/O reference](references/implicit_io.md) |
 | Observability and resilience | [Observability reference](references/observability.md) |
 | SQLModel-first modeling | [SQLModel integration reference](references/sqlmodel.md) |
 | CRUD repository and standalone functions | [Basic CRUD reference](references/crud.md) |
+| Test database selection and fixture data | [Database testing reference](references/testing.md) |
 
 ## Canonical Composition Pattern
 
-This example shows the ownership boundaries. Adapt state storage and dependency wiring to the application's conventions.
+The framework-independent primitives live in [engine lifecycle](references/engine.md), [session management](references/session.md), and [transaction boundaries](references/transactions.md). Their canonical FastAPI adaptation, including lifespan state and `Annotated` dependencies, lives in [FastAPI database integration](references/fastapi.md).
 
-```python
+For background work that outlives a request, inject the shared factory and create a new session inside that task instead of retaining the request's session.
-from contextlib import AsyncExitStack, asynccontextmanager
-from collections.abc import AsyncIterator
-
-from fastapi import FastAPI
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-
-@asynccontextmanager
-async def lifespan(app: FastAPI) -> AsyncIterator[None]:
-    async with AsyncExitStack() as stack:
-        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
-        yield
-
-
-async def get_session() -> AsyncIterator[AsyncSession]:
-   async with app.state.session_factory() as session:
-      yield session
-```
-
-For direct construction without `AsyncExitStack`, put `await engine.dispose()` in a `finally` block. For background work that outlives a request, create a new session inside that task instead of retaining the request's session.
 
 ## Explanation Procedure
 
@@ -173,6 +161,7 @@ When reviewing code, verify:
 - Relationship and deferred-column access cannot surprise the event loop with implicit I/O.
 - Pool and timeout settings are justified by deployment behavior.
 - Tests exercise rollback, cleanup, concurrency, and lifespan behavior where relevant.
+- Tests use a dedicated database target and preserve production session mechanics.
 
 ## Anti-Patterns to Flag
 

docs/skills/async-fastapi-sqlmodel/references/crud.md

@@ -5,12 +5,13 @@
     - [SQLModel update-data tutorial](https://sqlmodel.tiangolo.com/tutorial/fastapi/update-extra-data/)
     - [SQLModel select tutorial](https://sqlmodel.tiangolo.com/tutorial/select/)
     - [SQLAlchemy `AsyncSession` API](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.AsyncSession)
+    - [`nicegui-db` service functions](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/services/my_table.py)
 
 ??? abstract "Decision metadata"
     - Status: adopted
     - Decision level: advisory
     - Applies to: api-runtime, workers, tests
-    - Last reviewed: 2026-07-26
+    - Last reviewed: 2026-08-06
 
 ---
 
@@ -21,17 +22,17 @@ Show a small SQLModel CRUD layer in two forms:
 - independent functions for convenient standalone or composed operations;
 - a repository object that groups those functions behind one domain-oriented interface.
 
-Every public operation accepts an optional `AsyncSession`. When omitted, reads resolve the cached session factory and own a short-lived session, while writes resolve the same factory and own a complete session-and-transaction scope. When supplied, reads borrow the session and writes borrow its already-active caller-owned transaction. The repository stores configuration and delegates to the same functions without changing those semantics.
+Template-style public functions use `@with_session` and accept an optional `AsyncSession`. When the argument is omitted, the decorator resolves the cached session factory and owns a short-lived session. When supplied, the function borrows the session without controlling its lifetime or transaction. The decorator does not commit, so standalone writes need a visible transaction strategy; repository methods remain explicit-session operations for predictable composition.
 
 Use the same vocabulary at every layer:
 
 | Operation | Function | Repository method | Scope when session is omitted | Missing-row result |
 |---|---|---|---|---|
-| Create | `create_widget()` | `create()` | Owned transaction | Not applicable |
+| Create | `create_widget()` | `create()` | Owned session; no implicit commit | Not applicable |
 | Read one | `get_widget()` | `get()` | Owned session | `None` |
 | Read many | `list_widgets()` | `list()` | Owned session | Empty list |
-| Update | `update_widget()` | `update()` | Owned transaction | `None` |
+| Update | `update_widget()` | `update()` | Owned session; no implicit commit | `None` |
-| Delete | `delete_widget()` | `delete()` | Owned transaction | `None` |
+| Delete | `delete_widget()` | `delete()` | Owned session; no implicit commit | `None` |
 
 Functions and repository methods both put domain arguments first. Database configuration and sessions are keyword-only infrastructure arguments. This keeps call sites analogous and makes ownership choices visible.
 
@@ -58,159 +59,132 @@ This reference uses direct field arguments and full-update semantics to keep the
 
 ## Independent CRUD Functions
 
-Functions are the simplest default when grouping state or behavior in an object adds no value. Each function is a complete operation boundary: it can run standalone by resolving the cached factory from `database_url`, or compose into a caller-owned scope through `session`.
+Functions are the simplest default when grouping state or behavior in an object adds no value. Decorate public service functions when both standalone reads and explicit composition are useful. The assertion narrows the optional type after decorator injection and catches accidental explicit `None` calls.
 
 ```python
-from sqlalchemy.ext.asyncio import AsyncSession
 from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
 
-from .session import session_scope
+from .session import with_session
-from .session import transaction_scope
 
 
+@with_session
 async def create_widget(
     name: str,
     description: str | None = None,
     *,
-    database_url: str,
     session: AsyncSession | None = None,
 ) -> Widget:
-    async with transaction_scope(
+    assert session is not None, "Session must be provided by with_session decorator"
-        database_url=database_url,
+    widget = Widget(name=name, description=description)
-        session=session,
+    session.add(widget)
-    ) as active_session:
+    await session.flush()
-        widget = Widget(name=name, description=description)
+    return widget
-        active_session.add(widget)
-        await active_session.flush()
-        return widget
 
 
+@with_session
 async def get_widget(
     widget_id: int,
     *,
-    database_url: str,
     session: AsyncSession | None = None,
 ) -> Widget | None:
-    async with session_scope(
+    assert session is not None, "Session must be provided by with_session decorator"
-        database_url=database_url,
+    return await session.get(Widget, widget_id)
-        session=session,
-    ) as active_session:
-        return await active_session.get(Widget, widget_id)
 
 
+@with_session
 async def list_widgets(
     *,
-    database_url: str,
     offset: int = 0,
     limit: int = 100,
     session: AsyncSession | None = None,
 ) -> list[Widget]:
+    assert session is not None, "Session must be provided by with_session decorator"
     if offset < 0:
         raise ValueError("offset must be non-negative")
     if not 1 <= limit <= 100:
         raise ValueError("limit must be between 1 and 100")
 
-    async with session_scope(
+    statement = select(Widget).order_by(Widget.id).offset(offset).limit(limit)
-        database_url=database_url,
+    return list(await session.scalars(statement))
-        session=session,
-    ) as active_session:
-        statement = select(Widget).order_by(Widget.id).offset(offset).limit(limit)
-        return list(await active_session.scalars(statement))
 
 
+@with_session
 async def update_widget(
     widget_id: int,
     name: str,
     description: str | None,
     *,
-    database_url: str,
     session: AsyncSession | None = None,
 ) -> Widget | None:
-    async with transaction_scope(
+    assert session is not None, "Session must be provided by with_session decorator"
-        database_url=database_url,
+    widget = await session.get(Widget, widget_id)
-        session=session,
+    if widget is None:
-    ) as active_session:
+        return None
-        widget = await active_session.get(Widget, widget_id)
-        if widget is None:
-            return None
 
-        widget.name = name
+    widget.name = name
-        widget.description = description
+    widget.description = description
-        await active_session.flush()
+    await session.flush()
-        return widget
+    return widget
 
 
+@with_session
 async def delete_widget(
     widget_id: int,
     *,
-    database_url: str,
     session: AsyncSession | None = None,
 ) -> Widget | None:
-    async with transaction_scope(
+    assert session is not None, "Session must be provided by with_session decorator"
-        database_url=database_url,
+    widget = await session.get(Widget, widget_id)
-        session=session,
+    if widget is None:
-    ) as active_session:
+        return None
-        widget = await active_session.get(Widget, widget_id)
-        if widget is None:
-            return None
 
-        await active_session.delete(widget)
+    await session.delete(widget)
-        await active_session.flush()
+    await session.flush()
-        return widget
+    return widget
 ```
 
 Update and delete load the row through the same session that mutates it. This avoids accepting detached instances from an earlier standalone read and gives both operations an explicit `None` result that the application layer can map to a domain or HTTP error. Delete returns the loaded object for callers that need its values, but that object represents a row scheduled for deletion and must not be reused as persistent state. List operations validate their bounds and order by the primary key so pagination is deterministic. Add a unique tiebreaker whenever ordering by a non-unique field.
 
-`flush()` sends pending writes and populates ordinary generated primary keys. It does not itself commit. For a standalone write, the surrounding owned `transaction_scope()` commits after the function body succeeds. For a supplied session, the caller's outer transaction retains commit and rollback ownership. Use `await active_session.refresh(widget)` only when the operation deliberately needs database-generated state that was not returned during the flush; an unconditional refresh adds another query.
+`flush()` sends pending writes and populates ordinary generated primary keys. It does not itself commit. A decorated write called without a session will therefore roll back when its owned session closes unless the function explicitly commits. Prefer passing a transaction-scoped session so several writes compose atomically. Use `await session.refresh(widget)` only when the operation deliberately needs database-generated state that was not returned during the flush; an unconditional refresh adds another query.
 
 ---
 
 ## Repository Object
 
-A repository can provide a stable domain-facing interface when several callers need the same grouped operations. It stores repeatable database configuration, never a mutable session. Every method delegates to the analogous function and exposes the same optional-session contract.
+A repository can provide a stable domain-facing interface when several callers need the same grouped operations. It remains stateless here: every method requires a session and delegates to the analogous function.
 
 ```python
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel.ext.asyncio.session import AsyncSession
 
 class WidgetRepository:
-    def __init__(self, database_url: str) -> None:
-        self.database_url = database_url
-
     async def create(
         self,
+        session: AsyncSession,
         name: str,
         description: str | None = None,
-        *,
-        session: AsyncSession | None = None,
     ) -> Widget:
         return await create_widget(
             name,
             description,
-            database_url=self.database_url,
             session=session,
         )
 
     async def get(
         self,
+        session: AsyncSession,
         widget_id: int,
-        *,
-        session: AsyncSession | None = None,
     ) -> Widget | None:
-        return await get_widget(
+        return await get_widget(widget_id, session=session)
-            widget_id,
-            database_url=self.database_url,
-            session=session,
-        )
 
     async def list(
         self,
+        session: AsyncSession,
         *,
         offset: int = 0,
         limit: int = 100,
-        session: AsyncSession | None = None,
     ) -> list[Widget]:
         return await list_widgets(
-            database_url=self.database_url,
             offset=offset,
             limit=limit,
             session=session,
@@ -218,34 +192,27 @@ class WidgetRepository:
 
     async def update(
         self,
+        session: AsyncSession,
         widget_id: int,
         name: str,
         description: str | None,
-        *,
-        session: AsyncSession | None = None,
     ) -> Widget | None:
         return await update_widget(
             widget_id,
             name,
             description,
-            database_url=self.database_url,
             session=session,
         )
 
     async def delete(
         self,
+        session: AsyncSession,
         widget_id: int,
-        *,
-        session: AsyncSession | None = None,
     ) -> Widget | None:
-        return await delete_widget(
+        return await delete_widget(widget_id, session=session)
-            widget_id,
-            database_url=self.database_url,
-            session=session,
-        )
 ```
 
-The object is intentionally thin. Tests can construct it with a test database URL or pass a transaction-scoped test session to individual methods. A caller-provided session always wins and remains open after the method returns. A standalone operation closes its owned session before returning, so returned objects are detached; load every required scalar, deferred column, and relationship explicitly before the scope exits, and do not mutate those objects expecting persistence.
+The object is intentionally thin. Tests pass a transaction-scoped test session directly. The caller always owns that session and its transaction, and the repository never closes or commits it.
 
 If a read participates in a later write, pass the same session and place both operations inside the explicit transaction. This avoids splitting one use case across sessions and keeps SQLAlchemy's autobegin behavior from obscuring transaction ownership. Add a repository only when its naming, shared query policy, dependency substitution, or domain boundary improves the application. Independent functions remain a valid and often clearer design.
 
@@ -253,12 +220,10 @@ If a read participates in a later write, pass the same session and place both op
 
 ## Transaction Ownership
 
-Compose multiple calls under one use-case transaction. At this boundary, a supplied session joins its already-active caller-owned transaction, while omitting the session creates a standalone session and transaction. Each nested CRUD write receives `active_session`, detects that transaction, and borrows it instead of committing independently.
+Compose multiple calls under one use-case transaction. `db_transaction_scope()` owns the standalone engine, factory, session, and transaction lifetimes. Decorated CRUD functions detect the supplied session and borrow it; repository methods receive it directly.
-
-The scope names describe exactly what they own: `session_scope()` manages session lifetime but never commits, while `transaction_scope()` manages a complete transaction only when it also creates the session. Both yield the name `active_session` because downstream CRUD code does not need to know whether the session was borrowed or owned.
 
 ```python
-from .session import transaction_scope
+from .session import db_transaction_scope
 
 
 async def replace_widget(
@@ -266,38 +231,33 @@ async def replace_widget(
     widget_id: int,
     replacement_name: str,
     replacement_description: str | None = None,
-    *,
-    session: AsyncSession | None = None,
 ) -> Widget | None:
-    async with transaction_scope(
+    async with db_transaction_scope() as active_session:
-        database_url=repository.database_url,
-        session=session,
-    ) as active_session:
         deleted_widget = await repository.delete(
+            active_session,
             widget_id,
-            session=active_session,
         )
         if deleted_widget is None:
             return None
 
         return await repository.create(
+            active_session,
             replacement_name,
             replacement_description,
-            session=active_session,
         )
 ```
 
-If creation fails, deletion rolls back with it. For a caller-owned transaction, wrap the call in `async with session.begin():` and pass that session. For a standalone use case, omit the session; the outer `transaction_scope()` commits on successful exit, rolls back on exception, and closes its owned session. Do not add direct `commit()` calls to CRUD functions or repository methods because that prevents callers from composing several operations atomically. See [transaction boundaries](transactions.md) and [session management](session.md) for ownership details.
+If creation fails, deletion rolls back with it. Inside an already-running application, prefer `async with session_factory.begin()` or `async with session.begin()` over `db_transaction_scope()` so the application-owned engine and factory remain in use. Do not add direct `commit()` calls to CRUD functions or repository methods because that prevents callers from composing several operations atomically. See [transaction boundaries](transactions.md) and [session management](session.md) for ownership details.
 
 ---
 
 ## Anti-Patterns
 
 - Storing one mutable `AsyncSession` on a long-lived repository object.
-- Constructing ad hoc factories or sessions instead of resolving the cached factory through the scope helpers.
+- Creating sessions manually inside functions already using `@with_session`.
-- Using `session_scope()` for an optional write, which would close an owned session without committing.
+- Passing database configuration through every CRUD call instead of injecting a session at the data-access boundary.
-- Accepting a supplied session for a write without requiring an active caller-owned transaction.
+- Assuming decorator-owned write sessions commit on close.
-- Calling `commit()` or `rollback()` directly instead of expressing ownership through `transaction_scope()`.
+- Forwarding explicit `session=None` when decorator injection was intended.
 - Accepting unbounded list queries.
 - Accepting detached ORM instances for update or delete when an identifier can be resolved in the active session.
 - Accessing unloaded attributes after a standalone repository read has closed its owned session.
@@ -307,16 +267,16 @@ If creation fails, deletion rolls back with it. For a caller-owned transaction,
 ## Operational Checks
 
 - Every CRUD call receives a task-local `AsyncSession`.
-- Standalone reads resolve the cached factory by database URL and close their owned session.
+- Standalone reads create and close a session at the service or application boundary.
-- Standalone writes resolve the cached factory and own commit, rollback, and session cleanup through `transaction_scope()`.
+- Standalone reads may omit `session`; decorated writes receive a transaction-scoped session or explicitly own their commit policy.
-- Supplied write sessions already have an active caller-owned transaction.
+- Supplied write sessions remain caller-owned.
-- Each complete operation, service, or use-case boundary borrows an active transaction or owns a complete session-and-transaction scope.
+- Each complete write operation declares a visible transaction boundary.
 - List operations have pagination and deterministic ordering where required.
 - Update requires values for both mutable fields; passing `None` explicitly clears the nullable description.
 - Get, update, and delete use the same identifier and missing-row semantics.
-- Functions and repository methods use domain arguments first and keyword-only infrastructure arguments consistently.
+- Decorated functions accept an optional keyword-only session; repository methods require one explicitly.
-- Standalone reads load all state needed after their owned session closes.
+- Standalone service reads load all state needed after their owned session closes.
-- Repository objects hold configuration or policy, never request-scoped session state.
+- Repository objects hold query policy when useful, never database configuration or request-scoped session state.
 
 ---
 
@@ -330,5 +290,5 @@ If creation fails, deletion rolls back with it. For a caller-owned transaction,
 - Delete tests verify the returned row and its absence after commit.
 - Failure tests verify that a surrounding transaction rolls back all composed CRUD calls.
 - Optional-session read tests verify borrowed sessions remain open and owned sessions close without committing.
-- Optional-session write tests verify supplied transactions remain caller-owned and standalone transactions commit or roll back before closing.
+- Decorated write tests verify supplied transactions remain caller-owned and omitted sessions do not imply a commit.
 - Composition tests pass one active session through several CRUD calls and verify one atomic commit or rollback.

docs/skills/async-fastapi-sqlmodel/references/engine.md

@@ -1,45 +1,92 @@
 # Async SQLAlchemy Engine
 
 !!! info "Primary sources"
+    - [Python `asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager)
     - [Python `functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache)
     - [SQLAlchemy connections](https://docs.sqlalchemy.org/en/21/core/connections.html)
     - [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
     - [SQLAlchemy pooling and multiprocessing](https://docs.sqlalchemy.org/en/21/core/pooling.html#pooling-multiprocessing)
-    - [FastAPI lifespan events](https://fastapi.tiangolo.com/advanced/events/)
+    - [SQLAlchemy SQLite transaction control](https://docs.sqlalchemy.org/en/21/dialects/sqlite.html#enabling-non-legacy-sqlite-transactional-modes-with-the-sqlite3-or-aiosqlite-driver)
+    - [SQLAlchemy SQLite foreign-key support](https://docs.sqlalchemy.org/en/21/dialects/sqlite.html#foreign-key-support)
+    - [SQLite PRAGMA reference](https://www.sqlite.org/pragma.html)
+    - [`nicegui-db` engine implementation](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/db/engine.py)
 
 ---
 
 ## Engine Ownership Model
 
-Create one async engine per process per database URL and keep engine construction independent from FastAPI.
+Resolve one async engine for each database URL within an application, worker, command, or test lifecycle.
 
-- SQLAlchemy guidance: the engine is intended as a long-lived, concurrent registry over pooled DB connections, not a per-request object.
+- SQLAlchemy guidance: the engine is intended as a long-lived, concurrent registry over pooled DB connections, not a per-operation object.
-- A cached function provides stable process-local engine identity without making framework state the only way to obtain it.
+- `get_engine(database_url)` owns URL-keyed engine construction and caching.
-- FastAPI lifespan starts and stops that independently defined resource; it does not contain the construction policy.
+- The composition root enters `engine_scope(database_url)` once and therefore owns initialization and disposal.
+- Services and repositories receive a session or session factory; they do not resolve an engine.
 
 !!! tip "Practical rule"
-    - Exactly one `create_async_engine(...)` call in the cached engine factory.
+    - Exactly one cached engine for each database URL during an active application-owned lifecycle.
-    - Zero `create_async_engine(...)` calls in request handlers.
+    - Exactly one active owning `engine_scope()` for a given URL.
-    - Zero calls to the cached factory from repository code.
+    - Zero `create_async_engine(...)` calls in feature code.
+    - Zero engine lookup or disposal calls in repository code.
 
 ---
 
-## Cached Engine Factory
+## Cached Engine Resolution
 
-Use [`functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache) on a synchronous factory. Creating an `AsyncEngine` configures the dialect and pool; it does not need to await a database connection.
+[`functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache) makes the database URL the engine identity:
 
 ```python
 from functools import cache
 
-from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import create_async_engine
 
 
 @cache
 def get_engine(database_url: str) -> AsyncEngine:
-    return create_async_engine(
+    engine = create_async_engine(database_url, pool_pre_ping=True)
-        database_url,
+    if engine.dialect.name == "sqlite":
-        pool_pre_ping=True,
+        configure_aiosqlite_engine(engine)
-    )
+    return engine
+```
+
+Repeated calls with the same exact URL return the same `AsyncEngine`; different URLs produce independent cache entries. Construction configures the dialect and pool but normally does not open a database connection until the first operation. SQLite event listeners are installed only when a new cached engine is constructed, before its first connection.
+
+Resolve settings into the final URL before calling `get_engine()`. Services and repositories should not call it directly: the cache controls construction identity, not ownership.
+
+## Owning Engine Scope
+
+Use one [`asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager) to pair cached resolution and optional schema initialization with disposal:
+
+```python
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlmodel import SQLModel
+
+
+@asynccontextmanager
+async def engine_scope(
+    database_url: str,
+    *,
+    initialize: bool = True,
+) -> AsyncGenerator[AsyncEngine]:
+    engine = get_engine(database_url)
+    if initialize:
+        await initialize_db(database_url)
+
+    try:
+        yield engine
+    finally:
+        await dispose_engine(database_url)
+
+
+async def initialize_db(database_url: str) -> None:
+    from . import models  # noqa: F401
+
+    engine = get_engine(database_url)
+    async with engine.begin() as connection:
+        await connection.run_sync(SQLModel.metadata.create_all)
 
 
 async def dispose_engine(database_url: str) -> None:
@@ -48,52 +95,28 @@ async def dispose_engine(database_url: str) -> None:
         await engine.dispose()
     finally:
         get_engine.cache_clear()
-
-
-async def refresh_engine(database_url: str) -> AsyncEngine:
-    await dispose_engine(database_url)
-    return get_engine(database_url)
 ```
 
-The database URL is an explicit, hashable cache key. Calls with the same URL return the same engine; a different URL receives a different engine. If engine options vary at runtime, make them explicit hashable arguments too.
+The code that enters `engine_scope()` owns the engine. It keeps that scope open for the complete application, worker, command, or test lifecycle and passes the yielded engine into session-factory construction. Successful and exceptional exits both dispose the pool and invalidate cached engine resolution.
 
-Resolve settings at the composition boundary and call `get_engine(settings.database_url)`. Do not hide settings lookup or engine creation inside feature code.
+Initialization imports the model package so every table is registered, then runs `SQLModel.metadata.create_all()` in `engine.begin()`. This is suitable for the template and focused tests. Use migrations instead when schema evolution is part of the deployment contract. Pass `initialize=False` only when another owner provisions the schema or a test is directly exercising construction without schema setup.
 
-## Thin FastAPI Lifespan Wrapper
+`dispose_engine()` clears the complete function cache, not only the requested URL. This matches the template and is safe under its intended single-database lifecycle. Applications that own several simultaneously active database URLs need per-key lifecycle management rather than this global invalidation behavior.
 
-The lifespan context manager only connects the cached resource to FastAPI ownership:
+Workers, scripts, and other composition roots enter `database_scope()` directly:
 
 ```python
-from collections.abc import AsyncIterator
+async with database_scope(settings.database_url) as session_factory:
-from contextlib import asynccontextmanager
+    await run_worker(session_factory)
-
-from fastapi import FastAPI
-
-
-@asynccontextmanager
-async def lifespan(app: FastAPI) -> AsyncIterator[None]:
-    database_url = app.state.settings.database_url
-    engine = get_engine(database_url)
-    app.state.engine = engine
-
-    try:
-        yield
-    finally:
-        await dispose_engine(database_url)
-
-
-app = FastAPI(lifespan=lifespan)
 ```
 
-`dispose()` closes checked-in connections and replaces the pool, but it does not remove the Python object from `functools.cache`. `dispose_engine()` clears the cache even if driver cleanup raises, preventing a later lifespan run or test from retrieving that engine instance.
+`database_scope()` is defined in [session management](session.md). It enters `engine_scope()` and creates the factory bound to the yielded engine.
 
-This simple cleanup assumes one configured database URL per process. If a process intentionally owns several cached engines, use a small registry with per-key removal instead of clearing the whole cache. For a fixed engine, `try/finally` is sufficient; use `AsyncExitStack` when lifespan composes multiple conditional or dynamically acquired resources.
+Do not overlap two owning scopes for the same URL. Both resolve the same cached engine, and the first scope to exit disposes it and clears the cache while the other still refers to it. For several fixed databases, use one non-overlapping owner per URL and account for global cache invalidation; use [`AsyncExitStack`](https://docs.python.org/3/library/contextlib.html#contextlib.AsyncExitStack) only after adopting lifecycle semantics that support several simultaneous owners.
 
-When directly testing engine construction or lifespan behavior:
+When directly testing engine construction or lifecycle behavior, enter `engine_scope()` in the test or fixture. Exiting the context disposes the engine even when the test fails and clears the cache for the next lifecycle.
 
-- Call `get_engine.cache_clear()` before the test to remove process-local state.
+See [FastAPI database integration](fastapi.md) for adapting `database_scope()` to application lifespan and dependency injection.
-- Dispose any engine the test creates.
-- Clear the cache again during teardown, even when the test fails.
 
 ---
 
@@ -110,6 +133,62 @@ Use SQLAlchemy async driver URLs:
 
 ---
 
+## SQLite Connection and Transaction Policy
+
+SQLite settings do not form one indivisible bundle:
+
+- `PRAGMA foreign_keys=ON` is a correctness requirement when the schema declares foreign keys. SQLite requires it on every connection, including the connection used by `metadata.create_all()`.
+- Disabling the driver's implicit `BEGIN` and emitting `BEGIN` from SQLAlchemy provides non-legacy transaction behavior for `aiosqlite`. This makes SELECT, DDL, and SAVEPOINT behavior participate in SQLAlchemy's transaction boundary consistently.
+- `PRAGMA busy_timeout` is a per-connection lock-wait policy. Choose the duration from the application's latency and contention requirements.
+- `PRAGMA journal_mode=WAL` is an optional file-database concurrency policy. WAL persists in the database file, cannot be enabled for an in-memory database, and is not a substitute for transaction control.
+
+Install instance-level listeners exactly once, immediately after constructing an `aiosqlite` engine and before its first connection:
+
+```python
+from sqlalchemy import event
+from sqlalchemy.engine import Connection
+from sqlalchemy.engine.interfaces import DBAPIConnection
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+
+def configure_aiosqlite_engine(
+    engine: AsyncEngine,
+    *,
+    busy_timeout_ms: int | None = 30_000,
+    enable_wal: bool = False,
+) -> None:
+    if engine.dialect.name != "sqlite" or engine.dialect.driver != "aiosqlite":
+        raise ValueError("Expected a sqlite+aiosqlite engine")
+    if busy_timeout_ms is not None and busy_timeout_ms < 0:
+        raise ValueError("busy_timeout_ms must be non-negative")
+
+    @event.listens_for(engine.sync_engine, "connect")
+    def configure_connection(dbapi_connection: DBAPIConnection, _: object) -> None:
+        dbapi_connection.isolation_level = None
+        cursor = dbapi_connection.cursor()
+        try:
+            cursor.execute("PRAGMA foreign_keys=ON")
+            if busy_timeout_ms is not None:
+                cursor.execute(f"PRAGMA busy_timeout={busy_timeout_ms}")
+            if enable_wal:
+                cursor.execute("PRAGMA journal_mode=WAL")
+                journal_mode = cursor.fetchone()
+                if journal_mode is None or journal_mode[0].lower() != "wal":
+                    raise RuntimeError("SQLite could not enable WAL mode")
+        finally:
+            cursor.close()
+
+    @event.listens_for(engine.sync_engine, "begin")
+    def begin_transaction(connection: Connection) -> None:
+        connection.exec_driver_sql("BEGIN")
+```
+
+The `connect` listener receives the adapted synchronous DBAPI connection exposed by `engine.sync_engine`; event callbacks themselves are synchronous even though application queries use the async engine. Setting `isolation_level=None` and adding the `begin` listener are one transaction-control strategy and must remain paired. Do not combine this pair with SQLAlchemy's driver-level `AUTOCOMMIT` isolation mode.
+
+The default above enables foreign keys and modern transaction boundaries for file and in-memory databases. Enable WAL only for a file-backed database after deciding that its read/write concurrency model is appropriate. Treat `30_000` as an example policy, not a universal default; `connect_args={"timeout": 30.0}` at engine construction is another way to configure the underlying SQLite lock timeout.
+
+---
+
 ## Pooling Defaults and Tuning
 
 Default behavior is usually correct first:
@@ -123,14 +202,36 @@ 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.
 
+### When `StaticPool` Is Appropriate
+
+Use [`StaticPool`](https://docs.sqlalchemy.org/en/21/core/pooling.html#sqlalchemy.pool.StaticPool) only when every checkout must reuse one DBAPI connection and all database access is serialized. Typical cases are:
+
+- A serial test suite using a private in-memory SQLite database. The `sqlite+aiosqlite://` URL already selects `StaticPool` automatically, so specifying `poolclass=StaticPool` is normally redundant.
+- A narrowly scoped SQLite engine that must preserve connection-local state, such as temporary tables, across SQLAlchemy connection or session checkouts.
+
+When explicit configuration is required:
+
+```python
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.pool import StaticPool
+
+engine = create_async_engine(
+    "sqlite+aiosqlite:///./test.db",
+    poolclass=StaticPool,
+)
+```
+
+`StaticPool` is not a general performance optimization or a way to make SQLite concurrent. All sessions share one underlying connection and its single transaction state, so one session's `COMMIT` or `ROLLBACK` can interfere with another session. Do not use it when several sessions or tasks may access the engine concurrently. For concurrent in-memory work, use a named shared-cache SQLite URL so pooled connections have independent transaction state, or use a temporary file database. See [SQLite test targets](testing.md#sqlite-targets) for those patterns.
+
 ---
 
 ## Disposal Semantics
 
-`engine.dispose()` replaces/disposes the pool, but only checked-in connections are immediately closed.
+`dispose_engine(database_url)` resolves the cached engine, awaits `engine.dispose()`, and clears the engine cache in a `finally` block. `engine.dispose()` replaces/disposes the pool, but only checked-in connections are immediately closed.
 
 Rules:
 - Dispose when the app is shutting down.
+- Clear cached resolution even when disposal raises, so a later lifecycle cannot receive the failed engine object.
 - 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.
 
@@ -151,21 +252,30 @@ This prevents broken socket state and cross-process connection corruption.
 
 ## What Not to Do
 
-- Create an engine inside every request dependency.
+- Create an engine inside each operation or unit of work.
 - Create/dispose engines inside repository methods.
-- Call `get_engine()` from repositories instead of injecting their engine or session dependency.
+- Resolve an engine from repositories instead of injecting a session dependency.
 - Keep engine creation as a hidden side effect of import-time module globals.
-- Dispose a cached engine without clearing the cache during final teardown.
+- Keep a session factory alive after its bound engine scope exits.
-- Use deprecated FastAPI startup/shutdown events together with lifespan.
+- Enter overlapping engine scopes for the same cached URL.
+- Treat `cache_clear()` as per-URL invalidation when it clears every cached engine.
+- Use `metadata.create_all()` as a substitute for required production migrations.
+- Install the same SQLite event listeners more than once on one engine.
+- Enable WAL blindly for in-memory SQLite or treat a busy timeout as a concurrency guarantee.
 
 ---
 
 ## Engine Design Checklist
 
-- One engine per process per DB URL.
+- One cached engine per exact database URL during an active lifecycle.
-- Engine created by one cached, framework-independent factory.
+- One owning engine scope per URL, with no overlapping owners.
-- Lifespan only retrieves, exposes, disposes, and uncaches the engine.
+- Cached resolution, optional initialization, disposal, and cache invalidation follow one framework-independent lifecycle.
+- The composition root enters the database scope once and keeps it open until shutdown.
+- Session factory created inside, and never outlives, its engine scope.
+- Model registration occurs before `metadata.create_all()` when initialization is enabled.
 - Async driver URL matches backend (`asyncpg` or `aiosqlite`).
+- `aiosqlite` foreign-key and transaction listeners installed once before first use.
+- WAL enabled only as an explicit policy for a file-backed SQLite database.
 - Pooling strategy is explicit for non-default needs.
-- No request-path engine creation.
+- No feature-path engine creation.
-- Tests dispose engines and clear cached state deterministically.
+- Tests enter the same scope and receive deterministic disposal plus cache cleanup.

docs/skills/async-fastapi-sqlmodel/references/fastapi.md

@@ -0,0 +1,192 @@
+# FastAPI Database Integration
+
+!!! info "Primary sources"
+    - [FastAPI lifespan events](https://fastapi.tiangolo.com/advanced/events/)
+    - [FastAPI dependencies with `yield`](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/)
+    - [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/)
+    - [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
+    - [`nicegui-db` application lifespan](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/ui/app.py)
+    - [`nicegui-db` database dependencies](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/ui/dependency.py)
+
+---
+
+## Purpose
+
+Connect the framework-independent database tools to FastAPI:
+
+- lifespan enters one application-owned `database_scope()`,
+- application state holds settings and the resulting session factory,
+- dependencies create one session per request,
+- `Annotated` aliases make route ownership concise and explicit.
+
+The underlying resource and transaction rules remain in [engine lifecycle](engine.md), [session management](session.md), and [transaction boundaries](transactions.md).
+
+---
+
+## Lifespan Ownership
+
+Enter `database_scope()` once for the complete application lifecycle. Store the session factory, not the engine, because request code needs sessions rather than direct pool access:
+
+```python
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from .config import Settings
+from .config import get_database_url
+from .db import database_scope
+
+
+@asynccontextmanager
+async def lifespan(settings: Settings, app: FastAPI) -> AsyncGenerator[None]:
+    app.state.settings = settings
+    db_url = get_database_url(settings)
+
+    try:
+        async with database_scope(db_url) as session_factory:
+            app.state.session_factory = session_factory
+            yield
+    finally:
+        del app.state.settings
+        del app.state.session_factory
+```
+
+The application factory binds `settings` to lifespan, for example with `partial(lifespan, settings)`. Lifespan does not construct resources per request. It enters the same framework-independent scope used by scripts, workers, and tests, keeps that scope open while requests are served, and lets it dispose the engine and clear cached engine resolution during shutdown.
+
+The template's unconditional `del app.state.session_factory` mirrors an expected successful startup. If `database_scope()` raises before assignment, cleanup can raise `AttributeError` and obscure the startup error. A production hardening option is to assign a sentinel before the `try` or delete conditionally; that changes failure behavior and is not part of the exact template mechanics.
+
+Only store the engine too when application-level code genuinely needs direct Core operations, pool instrumentation, or engine-specific diagnostics. Routes and repositories should normally receive an `AsyncSession`.
+
+---
+
+## Session Factory Dependency
+
+A synchronous dependency retrieves the already-created factory from application state:
+
+```python
+from typing import Annotated
+
+from fastapi import Depends
+from fastapi import Request
+
+from .session import SessionFactory
+
+
+def _get_session_factory(request: Request) -> SessionFactory:
+    return request.app.state.session_factory
+
+
+type SessionFactoryDep = Annotated[SessionFactory, Depends(_get_session_factory)]
+```
+
+`Depends()` does not create or cache a factory here. It only exposes the lifespan-owned object. This function is also the narrow seam that tests can override when they need a different factory.
+
+---
+
+## Request Session Dependencies
+
+Use a session-only dependency for reads and other request conversations that must not commit implicitly:
+
+```python
+from collections.abc import AsyncGenerator
+
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+async def _get_session(session_factory: SessionFactoryDep) -> AsyncGenerator[AsyncSession]:
+    async with session_factory() as owned_session:
+        yield owned_session
+
+
+type SessionDep = Annotated[AsyncSession, Depends(_get_session)]
+```
+
+The dependency creates and closes one session per request. Closing rolls back any unfinished autobegun transaction; it does not commit.
+
+---
+
+## Route Usage
+
+Read route:
+
+```python
+@router.get("/items/{item_id}")
+async def get_item(item_id: int, session: SessionDep) -> Item | None:
+    return await find_item(session, item_id)
+```
+
+Write route:
+
+```python
+@router.post("/items")
+async def create_item(payload: ItemCreate, session: SessionDep) -> Item:
+    async with session.begin():
+        return await insert_item(session, payload)
+```
+
+The template exposes only `SessionDep`; it does not hide commit behavior in dependency teardown. Choose one visible write convention per application:
+
+- place `async with session.begin():` around a complete write unit, which commits on success and rolls back on exception; or
+- call `await session.commit()` explicitly after all writes when the route is the complete unit, as the template's simple UI action does.
+
+The context-manager form scales better to several statements and makes exception rollback visible. Direct `commit()` is concise but requires the route to preserve the single-commit invariant and handle any recovery needs. Do not combine both conventions in one route. Lower-level data-access functions receive the existing session and remain unaware of FastAPI.
+
+---
+
+## Background Work
+
+A request session belongs to that request and must not be retained by a background task. Inject or otherwise provide the application session factory, then create a new session inside the task:
+
+```python
+async def run_background_job(session_factory: SessionFactory) -> None:
+    async with session_factory.begin() as session:
+        await process_pending_items(session)
+```
+
+If work must survive application shutdown, it needs an independently owned worker lifecycle rather than the FastAPI lifespan-owned factory.
+
+---
+
+## Testing and Overrides
+
+Override the narrow dependency that matches the test objective:
+
+- Override `_get_session_factory` to preserve production request-session behavior with a test factory.
+- Override `_get_session` when a test must inject one transaction-scoped session directly.
+- Verify each lifespan receives a fresh engine and session factory and removes application state during teardown.
+- Remove overrides during teardown so mutable application state does not leak between tests.
+
+```python
+app.dependency_overrides[_get_session] = get_test_session
+try:
+    yield app
+finally:
+    app.dependency_overrides.pop(_get_session, None)
+```
+
+See [database testing](testing.md) for outer transactions, SAVEPOINT-backed fixtures, and database target selection.
+
+---
+
+## Anti-Patterns
+
+- Creating an engine or session factory in a request dependency.
+- Reading settings and constructing database resources from repositories.
+- Storing one mutable `AsyncSession` on `app.state`.
+- Sharing a request session with concurrent or background tasks.
+- Assuming `SessionDep` commits when dependency cleanup runs.
+- Keeping `app.state.session_factory` after its `database_scope()` exits.
+- Using deprecated startup and shutdown event handlers alongside lifespan.
+
+---
+
+## Integration Checklist
+
+- Lifespan enters exactly one `database_scope()` for each application lifecycle.
+- Application state stores the yielded session factory.
+- Session dependencies create and close one session per request.
+- The session dependency owns request session closure but not commit behavior.
+- Routes use `Annotated` aliases and receive sessions, not engines.
+- Background tasks create their own sessions from a still-live factory.
+- Tests override and restore dependencies deterministically.

docs/skills/async-fastapi-sqlmodel/references/index.md

@@ -8,13 +8,15 @@ Purpose: concept registry for the principles, mechanics, and implementation guid
 
 | Concept | File | Status | Decision Level | Owner | Last Reviewed |
 |---|---|---|---|---|---|
-| Engine lifecycle and ownership | [engine.md](engine.md) | adopted | mandatory | platform/backend | 2026-06-17 |
+| Engine lifecycle and ownership | [engine.md](engine.md) | adopted | mandatory | platform/backend | 2026-08-06 |
-| Session factory and scope | [session.md](session.md) | adopted | mandatory | platform/backend | 2026-06-17 |
+| Session factory and scope | [session.md](session.md) | adopted | mandatory | platform/backend | 2026-08-06 |
+| FastAPI lifespan and dependency injection | [fastapi.md](fastapi.md) | adopted | mandatory | platform/backend | 2026-08-06 |
 | Transaction boundaries | [transactions.md](transactions.md) | adopted | mandatory | platform/backend | 2026-06-17 |
 | Implicit ORM I/O under asyncio | [implicit_io.md](implicit_io.md) | adopted | advisory | platform/backend | 2026-06-17 |
 | Observability and resilience | [observability.md](observability.md) | adopted | mandatory | platform/backend | 2026-06-17 |
-| SQLModel modeling and async boundaries | [sqlmodel.md](sqlmodel.md) | adopted | mandatory | platform/backend | 2026-07-26 |
+| SQLModel modeling and async boundaries | [sqlmodel.md](sqlmodel.md) | adopted | mandatory | platform/backend | 2026-08-06 |
-| Basic CRUD repository and functions | [crud.md](crud.md) | adopted | advisory | platform/backend | 2026-07-26 |
+| Basic CRUD repository and functions | [crud.md](crud.md) | adopted | advisory | platform/backend | 2026-08-06 |
+| Test database targets and fixture data | [testing.md](testing.md) | adopted | mandatory | platform/backend | 2026-08-06 |
 
 ---
 

docs/skills/async-fastapi-sqlmodel/references/session.md

@@ -1,42 +1,46 @@
 # Async SQLAlchemy Session Management
 
 !!! info "Primary sources"
-    - [Python `functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache)
     - [Python `asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager)
+    - [Python `functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache)
+    - [Python `inspect.signature`](https://docs.python.org/3/library/inspect.html#inspect.signature)
     - [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
     - [SQLAlchemy session basics](https://docs.sqlalchemy.org/en/21/orm/session_basics.html)
-    - [FastAPI dependencies with yield](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/)
+    - [`nicegui-db` session implementation](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/db/session.py)
 
 ---
 
 ## Purpose
 
-Define one canonical session model for FastAPI + SQLAlchemy asyncio:
+Define one canonical session model for SQLAlchemy asyncio:
 
-- configure one shared session factory,
+- configure a lifespan-owned factory or resolve a URL-keyed cached factory,
-- create one AsyncSession per request or per unit-of-work,
+- create one AsyncSession per task or unit of work,
+- let callers supply a session when they already own the scope,
 - never share one AsyncSession across concurrent tasks.
 
 ---
 
 ## Scope and Non-Goals
 
-- In scope: session factory creation, FastAPI dependency wiring, request/task scoping, transaction demarcation.
+- In scope: session factory creation, task scoping, and transaction demarcation.
-- Out of scope: ORM model design, query optimization strategy, schema migration tooling.
+- Out of scope: framework dependency wiring, ORM model design, query optimization strategy, and schema migration tooling.
 
 ---
 
 ## Rules
 
-- Create one cached `async_sessionmaker` per app-owned AsyncEngine.
+- Create the application `async_sessionmaker` inside `database_scope()` and store it in application state for request dependencies.
-- Let repositories resolve the cached maker by database URL.
+- Use `get_session_factory(db_url)` and `resolve_session_factory()` for standalone decorated operations that do not receive the application factory.
-- Use a fresh AsyncSession for each request or explicit unit-of-work.
+- Use a fresh AsyncSession for each task or explicit unit of work.
-- Pass an `AsyncSession` directly to data-access functions.
+- Let reusable service functions accept `AsyncSession | None` and apply `@with_session` when standalone invocation is useful.
-- Borrow a caller-provided session without closing or committing it.
+- Pass an `AsyncSession` directly when composing several calls under one caller-owned scope.
+- Borrow a caller-provided session without beginning, closing, committing, or rolling it back.
 - Do not share AsyncSession across `asyncio.gather()` or parallel tasks.
 - Prefer direct dependency injection over global scoped-session patterns in new code.
 - Use explicit transaction boundaries (`async with session.begin():`) for writes.
-- When a use case accepts an optional session, borrow only an active caller-owned transaction or own the complete session-and-transaction scope.
+- Use `db_transaction_scope()` when a standalone operation must own engine, factory, session, and transaction lifetimes together.
+- Use `begin_nested()` directly and only when partial rollback through a database SAVEPOINT is required.
 
 ---
 
@@ -46,7 +50,7 @@ A session and a transaction solve related but different problems:
 
 | Concept | Responsibility | Typical lifetime |
 | --- | --- | --- |
-| `AsyncSession` | Provides the ORM workspace: executes queries, tracks loaded and changed objects in its identity map, and flushes pending changes. It also coordinates access to a database connection. | One request, task, or explicit unit of work. |
+| `AsyncSession` | Provides the ORM workspace: executes queries, tracks loaded and changed objects in its identity map, and flushes pending changes. It also coordinates access to a database connection. | One task or explicit unit of work. |
 | Transaction | Defines the atomic database boundary: all work inside it commits together on success or rolls back together on failure. | One complete operation that must have a single outcome. |
 
 A transaction belongs to a session; it is not an alternative to one. The session is the interface used by application and data-access code, while the transaction determines when that work becomes permanent. A session may coordinate sequential transactions during its lifetime, although short-lived application scopes commonly use one session for one transaction.
@@ -76,32 +80,68 @@ For most read-only operations, a session context is sufficient. Use an explicit
 
 An `async_sessionmaker[AsyncSession]` is a reusable configuration object and callable session producer. It stores how sessions should be created, including the engine binding and options such as `expire_on_commit=False`. It is not itself a session, connection, or transaction, and calling it does not make a shared global `AsyncSession`.
 
-Cache it by the application-owned engine so repeated composition calls return the same maker:
+The template exposes two construction paths with the same session options.
+
+The application-owned path creates a factory inside the engine lifecycle:
+
+```python
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from .engine import engine_scope
+
+type SessionFactory = async_sessionmaker[AsyncSession]
+
+
+@asynccontextmanager
+async def database_scope(
+    db_url: str,
+    *,
+    auto_flush: bool = True,
+) -> AsyncGenerator[SessionFactory]:
+    async with engine_scope(db_url) as engine:
+        yield async_sessionmaker(
+            bind=engine,
+            class_=AsyncSession,
+            expire_on_commit=False,
+            autoflush=auto_flush,
+        )
+```
+
+FastAPI lifespan enters this path once and stores the yielded factory on application state. The factory must not outlive the scope because its bound engine is disposed on exit.
+
+The standalone path caches a factory by URL and `auto_flush` policy:
 
 ```python
 from functools import cache
 
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
-
-from .engine import dispose_engine
 from .engine import get_engine
 
 
 @cache
-def get_session_factory(database_url: str) -> async_sessionmaker[AsyncSession]:
+def get_session_factory(
+    db_url: str,
+    *,
+    auto_flush: bool = True,
+) -> SessionFactory:
     return async_sessionmaker(
-        bind=get_engine(database_url),
+        bind=get_engine(db_url),
         class_=AsyncSession,
         expire_on_commit=False,
+        autoflush=auto_flush,
     )
 
 
-async def dispose_session_factory(database_url: str) -> None:
+def resolve_session_factory(settings: Settings | None = None) -> SessionFactory:
-    get_session_factory.cache_clear()
+    settings = settings or get_settings()
-    await dispose_engine(database_url)
+    db_url = get_database_url(settings)
+    return get_session_factory(db_url)
 ```
 
-`functools.cache` caches by argument equality and requires hashable arguments. The database URL is an explicit string key shared with the cached engine factory. The cache retains the returned maker until `get_session_factory.cache_clear()` runs. Cache the synchronous maker function, never an async function and never a produced `AsyncSession`.
+This path lets framework-independent helpers resolve one stable factory without receiving it through every call. The tradeoff is hidden configuration resolution and a second lifecycle mechanism. `dispose_engine()` clears `get_engine`'s cache but does not clear `get_session_factory`'s cache in the template. A cached factory remains bound to the disposed engine object; SQLAlchemy can create a new pool when that engine is used again, but a later `database_scope()` for the same URL can own a different engine. Treat cached standalone resolution as process-lifetime convenience, avoid repeated application lifecycles in one process, and clear both caches together if the template evolves to support them.
 
 Each call to `session_factory()` creates a distinct `AsyncSession`. The caller that invokes the factory owns that session lifetime and must close it, normally with `async with`:
 
@@ -110,212 +150,189 @@ async with session_factory() as session:
     ...
 ```
 
-The factory can be shared across requests and tasks. Sessions produced by it cannot be shared across concurrent tasks.
+The factory can be shared across operations and tasks. Sessions produced by it cannot be shared across concurrent tasks.
 
-An `async_sessionmaker` has no connection pool or async `dispose()` method of its own. `dispose_session_factory()` means "invalidate the cached maker, then dispose its engine." Clearing the maker first ensures no subsequent composition call can retrieve a maker bound to the engine being shut down.
+Passing the application factory directly has three useful consequences:
 
-Use the helper when shutting down or replacing the database resources:
+- Lower layers do not resolve settings or global resources.
-
+- Tests can inject a test factory directly through `session_scope(session_factory=...)` or FastAPI state.
-```python
+- Transaction ownership remains independent of engine construction.
-await dispose_session_factory(database_url)
-```
-
-Otherwise, a later call can return a maker that still references the old engine object. This matters in lifespan tests, application restarts within one process, and test suites that replace engines.
 
 ---
 
-## Optional Session Ownership
+## Database and Convenience Scopes
 
-A small [`asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager) can make repository methods composable. It borrows an existing session when supplied; otherwise it creates and closes one from a supplied factory:
+The template provides three framework-independent context managers:
 
 ```python
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 
+@asynccontextmanager
+async def db_session_scope(
+    db_url: str | None = None,
+) -> AsyncGenerator[AsyncSession]:
+    db_url = db_url or resolve_database_url()
+    async with database_scope(db_url) as session_factory, session_factory() as session:
+        yield session
+
+
+@asynccontextmanager
+async def db_transaction_scope(
+    db_url: str | None = None,
+) -> AsyncGenerator[AsyncSession]:
+    db_url = db_url or resolve_database_url()
+    async with database_scope(db_url) as session_factory, session_factory.begin() as session:
+        yield session
+
+
 @asynccontextmanager
 async def session_scope(
     *,
-    database_url: str,
+    settings: Settings | None = None,
+    session_factory: SessionFactory | None = None,
     session: AsyncSession | None = None,
-) -> AsyncIterator[AsyncSession]:
+) -> AsyncGenerator[AsyncSession]:
     if session is not None:
         yield session
         return
 
-    async with get_session_factory(database_url)() as owned_session:
+    session_factory = session_factory or resolve_session_factory(settings=settings)
+    async with session_factory() as owned_session:
         yield owned_session
 ```
 
-The branch is intentionally explicit. Python's [`nullcontext`](https://docs.python.org/3/library/contextlib.html#contextlib.nullcontext) can express the same borrow-or-own idea, but the branch keeps ownership and typing obvious.
+`db_session_scope()` owns a complete temporary database lifecycle and a session but does not commit. `db_transaction_scope()` owns the same resources plus a root transaction that commits on successful exit and rolls back on exception. Both initialize the schema by default because `database_scope()` enters `engine_scope()` with its default `initialize=True`. They are appropriate for scripts, commands, and isolated operations, not per-request use inside an already-running application.
 
-This helper manages session lifetime only:
+`session_scope()` is the borrow-or-create helper. Its precedence is supplied session, supplied factory, then settings-based cached factory resolution. A supplied session remains entirely caller-owned; the helper does not require an active transaction and does not begin, commit, roll back, or close it. An owned session is closed on exit, and unfinished autobegun work rolls back.
 
-- It does not close, commit, or roll back a supplied session; the caller owns it.
+Passing `session=None` is the same as omitting the session for `session_scope()` and therefore creates a session. This differs from `with_session`, which tests whether the argument name was bound rather than whether its value is non-null.
-- It closes a session that it creates. Closing releases resources and rolls back an unfinished transaction; it does not commit.
-- It does not start a transaction. Put `session.begin()` at the use-case boundary.
-- A supplied session wins; the cached factory is not resolved.
-- Otherwise, `database_url` selects the cached factory returned by `get_session_factory()`.
 
-Do not turn this into an implicit unit-of-work helper that sometimes commits. Whether work joins an existing transaction or creates a new one must remain visible to the caller.
+## Signature-Aware Session Injection
+
+`with_session` allows one async function to support standalone calls and explicit composition:
+
+```python
+from collections.abc import Awaitable
+from collections.abc import Callable
+from functools import wraps
+from inspect import signature
+
+
+def with_session[**P, R](
+    func: Callable[P, Awaitable[R]],
+) -> Callable[P, Awaitable[R]]:
+    sig = signature(func)
+
+    @wraps(func)
+    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+        bound = sig.bind_partial(*args, **kwargs)
+
+        if "session" in bound.arguments:
+            return await func(*args, **kwargs)
+
+        async with resolve_session_factory()() as session:
+            bound.arguments["session"] = session
+            return await func(*bound.args, **bound.kwargs)
+
+    return wrapper
+```
+
+The function must be async and expose a parameter named exactly `session`. The decorator preserves metadata with `wraps()`, binds positional and keyword arguments through the original signature, and injects a fresh session only when the caller omitted that argument.
+
+The distinction between omitted and explicit `None` is deliberate in the implementation:
+
+- `await operation()` injects and owns a session.
+- `await operation(session=existing_session)` borrows the caller's session.
+- `await operation(None)` or `await operation(session=None)` forwards `None` without injection.
+
+The decorated function therefore types the parameter as `AsyncSession | None = None` but should assert or guard after decoration. Explicit `None` is not a request for injection. This preserves ordinary Python call binding, but it means wrappers or callers must omit the argument instead of forwarding a nullable value.
+
+`with_session` owns session lifetime only. It does not begin or commit a transaction, so it is naturally suited to reads. Decorated writes must either manage a visible transaction or be called with a session from `db_transaction_scope()` or another caller-owned transaction. Prefer explicit factory or session injection when lifecycle transparency and test substitution matter more than call-site convenience.
 
 ---
 
-## Optional Transaction Ownership
+## Function and Service Boundaries
 
-Use a separate context manager when a service or use-case function must support both a caller-owned transaction and a standalone transaction. A supplied session must already be inside a transaction; otherwise the helper creates a session and transaction together with `async_sessionmaker.begin()`:
+Template service functions support both standalone and composed use by combining `@with_session` with an optional parameter:
 
 ```python
-@asynccontextmanager
+from sqlmodel import func
-async def transaction_scope(
+from sqlmodel import select
-    *,
-    database_url: str,
-    session: AsyncSession | None = None,
-) -> AsyncIterator[AsyncSession]:
-    if session is not None:
-        if not session.in_transaction():
-            raise RuntimeError("A supplied session must have an active transaction")
-        yield session
-        return
 
-    session_factory = get_session_factory(database_url)
+
-    async with session_factory.begin() as owned_session:
+@with_session
-        yield owned_session
+async def count_items(session: AsyncSession | None = None) -> int:
+    assert session is not None, "Session must be provided by with_session decorator"
+    result = await session.exec(select(func.count()).select_from(Item))
+    return result.one()
 ```
 
-Here, `begin()` is intentionally called on the [`async_sessionmaker`](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.async_sessionmaker.begin), not on an existing `AsyncSession`. The related APIs have different ownership semantics:
+The standalone call injects and closes a session:
-
-- `session_factory()` creates a session whose lifetime the surrounding code must manage; it does not commit automatically.
-- `session_factory.begin()` creates a new session and transaction together, commits on successful exit or rolls back on exceptional exit, and then closes the session.
-- `session.begin()` manages a transaction on an existing session but does not own or close that session.
-
-The factory form is equivalent in ownership terms to creating a session and then entering that session's transaction:
 
 ```python
-async with session_factory() as owned_session:
+count = await count_items()
-    async with owned_session.begin():
-        yield owned_session
 ```
 
-This helper makes transaction ownership follow the same explicit borrow-or-own mechanics as session ownership:
+A larger use case passes one caller-owned session through several decorated functions:
-
-- A supplied session and its active transaction remain caller-owned. The helper does not commit, roll back, or close them.
-- Without a supplied session, the helper owns the session and transaction. Successful exit commits; exceptional exit rolls back; either path closes the session.
-- Use this helper only at a complete operation, service, or use-case boundary. A public CRUD function or repository method may be such a boundary when its optional-session contract explicitly states that omitting the session owns and commits one transaction. Never use it inside a lower-level session-required helper.
-- Do not silently begin a transaction on a supplied session. That would make commit ownership depend on hidden helper behavior.
-
-Callers that supply a session make their ownership visible with an outer transaction:
 
 ```python
-async with session_factory() as session:
+async with session_factory.begin() as session:
-    async with session.begin():
+    count = await count_items(session)
-        await run_use_case(..., session=session)
+    await create_item(payload, session=session)
 ```
 
-Standalone callers omit the session and let the use case own the complete unit of work:
+The decorator sees the bound `session` argument and leaves all ownership with the caller. It never creates a SAVEPOINT or nested transaction.
+
+For low-level helpers that should never resolve settings, require a non-optional session and leave them undecorated. Application service objects may store the immutable session factory, but they must not store a mutable session:
 
 ```python
-await run_use_case(...)
+class ItemService:
+    def __init__(self, session_factory: SessionFactory) -> None:
+        self.session_factory = session_factory
+
+    async def find(self, item_id: int) -> Item | None:
+        async with self.session_factory() as session:
+            return await find_item(session, item_id)
 ```
 
+Code that already owns a transaction should call the session-required function directly. Repositories should normally remain in that session-required layer; the service or use-case boundary owns standalone session creation. This avoids optional-session APIs spreading into every data-access function.
+
 ---
 
-## Repository and Function Boundaries
+## SAVEPOINTs and Partial Failure
 
-Pass the database URL to repository constructors. The repository stores repeatable database configuration, not mutable session state, and `session_scope()` resolves the cached factory when a standalone operation needs a session:
+Use [`begin_nested()`](https://docs.sqlalchemy.org/en/21/orm/session_transaction.html#using-savepoint) only when failure inside one portion of an operation should roll back that portion while preserving the outer transaction:
 
 ```python
-from sqlalchemy import select
+async with db_transaction_scope() as session:
-from sqlalchemy.ext.asyncio import AsyncSession
+    order = await insert_order(session, payload)
 
+    try:
+        async with session.begin_nested():
+            await apply_optional_discount(session, order)
+    except DiscountError:
+        pass
 
-async def find_item(session: AsyncSession, item_id: int) -> Item | None:
+    await reserve_inventory(session, order)
-    statement = select(Item).where(Item.id == item_id)
-    return await session.scalar(statement)
-
-
-class ItemRepository:
-    def __init__(self, database_url: str) -> None:
-        self.database_url = database_url
-
-    async def find(
-        self,
-        item_id: int,
-        *,
-        session: AsyncSession | None = None,
-    ) -> Item | None:
-        async with session_scope(
-            database_url=self.database_url,
-            session=session,
-        ) as active_session:
-            return await find_item(active_session, item_id)
 ```
 
-This split gives each layer one job:
+Important SAVEPOINT semantics:
 
-- The repository object identifies its database configuration and creates a session only for a standalone call.
+- `begin_nested()` starts a root transaction if one is not already active, so call it inside a visible outer transaction when that ownership matters.
-- Standalone calls reuse the cached factory selected by database URL.
+- Entering `begin_nested()` unconditionally flushes pending session state, regardless of the `autoflush` setting.
-- A caller can pass a session to join an existing unit of work; the repository borrows it.
+- Successful exit releases the SAVEPOINT; it does not commit the outer transaction.
-- The access function owns only the query and requires an existing `AsyncSession`.
+- Exceptional exit rolls back to the SAVEPOINT and leaves the outer transaction active.
-- Application wiring supplies the production factory.
+- In SQLAlchemy 2.x, `session.commit()` commits the outermost transaction. Never call it to release a SAVEPOINT; let the nested context manager manage its transaction handle.
-- Tests can use a test database URL or call `find_item()` with a transaction-scoped test session.
 
-When several repository operations must share one transaction, pass the same session through each call. Put the transaction at the use-case boundary:
+Do not create a SAVEPOINT merely because one service calls another. SAVEPOINTs add database work and alter flush and error-recovery behavior. Use them only for explicit partial-failure requirements such as skipping one conflicting row while retaining the rest of a batch.
-
-```python
-async with session_factory() as session:
-    async with session.begin():
-        item = await repository.find(item_id, session=session)
-        await update_item(session, item, changes)
-```
-
-This preserves atomicity without making repository objects hold mutable `AsyncSession` instances across calls.
 
 ---
 
-## Canonical FastAPI Dependency Pattern
+## Framework Integration
 
-```python
+Keep framework adapters outside these session primitives. See [FastAPI database integration](fastapi.md) for lifespan ownership, `Annotated` dependency aliases, and read-versus-write request sessions.
-from collections.abc import AsyncIterator
-
-from fastapi import Depends
-from fastapi import Request
-from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.ext.asyncio import async_sessionmaker
-
-
-type SessionFactory = async_sessionmaker[AsyncSession]
-
-
-def resolve_session_factory(request: Request) -> SessionFactory:
-    return get_session_factory(request.app.state.settings.database_url)
-
-
-async def get_db_session(
-    session_factory: SessionFactory = Depends(resolve_session_factory),
-) -> AsyncIterator[AsyncSession]:
-    async with session_factory() as session:
-        yield session
-```
-
-Route usage:
-
-```python
-from fastapi import APIRouter, Depends
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from .session import get_db_session
-
-router = APIRouter()
-
-
-@router.post("/items")
-async def create_item(session: AsyncSession = Depends(get_db_session)) -> dict:
-    async with session.begin():
-        # write operations here
-        ...
-    return {"status": "ok"}
-```
 
 ---
 
@@ -327,9 +344,9 @@ async def create_item(session: AsyncSession = Depends(get_db_session)) -> dict:
 
 ## SQLModel Alignment
 
-- Use SQLModel as the default model and statement layer while keeping the same session ownership model: one `async_sessionmaker`, one `AsyncSession` per request/unit-of-work.
+- Use SQLModel as the default model and statement layer while keeping the same session ownership model: one `async_sessionmaker`, one `AsyncSession` per task or unit of work.
 - SQLModel does not replace SQLAlchemy async lifecycle primitives; it provides model declaration, validation, and typing ergonomics on top of them.
-- Do not mix ad hoc session construction with the canonical async dependency.
+- Do not mix ad hoc session construction with the canonical session factory.
 
 ---
 
@@ -343,41 +360,45 @@ async def create_item(session: AsyncSession = Depends(get_db_session)) -> dict:
 
 ## Anti-Patterns
 
-- A singleton/global AsyncSession reused across requests.
+- A singleton/global AsyncSession reused across tasks or operations.
 - Sharing one AsyncSession across parallel tasks.
 - Passing an application-global AsyncSession to a repository constructor.
-- Caching an `AsyncSession` instead of caching `async_sessionmaker`.
+- Creating a new `async_sessionmaker` in each operation.
-- Leaving a cached maker pointing at a disposed or replaced engine.
+- Retaining a session factory after its bound engine scope exits.
-- Calling the session factory inside low-level access functions such as `find_item()`.
+- Using cached standalone factory resolution when the application factory is already available.
-- Hidden session creation in lower access functions with no caller control.
+- Assuming `with_session` starts or commits a transaction.
+- Forwarding `session=None` to a decorated function when injection was intended.
 - Closing or committing a session supplied by the caller.
-- Starting a new transaction inside a helper that may receive a session already in a transaction.
 - Silently starting or committing a transaction on a supplied session.
+- Creating a SAVEPOINT for ordinary nested service calls.
+- Hiding root transaction, joined transaction, and SAVEPOINT behavior behind one mode-driven `atomic_scope()` helper.
+- Calling `session.commit()` inside a SAVEPOINT scope.
 - Mixing commit/rollback ownership across layers without a declared boundary.
 
 ---
 
 ## Operational Checks
 
-- Exactly one cached `async_sessionmaker` exists per application engine.
+- The FastAPI application factory is created inside `database_scope()` and does not outlive its bound engine.
-- Session factory caches are cleared before their engines are disposed or replaced.
+- Cached standalone factories are used only where application-state injection is unavailable.
-- Request handlers receive sessions from one canonical dependency.
+- `session_scope()` precedence is supplied session, supplied factory, then settings-based resolution.
+- Decorated functions receive injection only when the `session` argument is omitted.
 - No code path creates AsyncSession in module import side effects.
-- Background jobs and API handlers each create task-local sessions.
+- Concurrent jobs and operations each create task-local sessions.
 
 ---
 
 ## Testing Checks
 
-- Repository constructors accept a test database URL without FastAPI startup.
+- Service constructors accept a test session factory without framework startup.
 - Session-taking access functions accept a transaction-scoped test session directly.
-- Optional-session tests verify that borrowed sessions remain open and created sessions close.
+- `session_scope()` tests cover supplied-session, supplied-factory, and settings-resolution precedence.
-- Optional-session tests verify that neither path commits implicitly.
+- `db_transaction_scope()` tests verify commit on success, rollback on failure, session closure, engine disposal, and cache cleanup.
-- Optional-transaction tests verify supplied sessions require an active transaction and remain caller-owned.
+- `with_session` tests cover omitted, positional, keyword, and explicit-`None` session arguments.
-- Optional-transaction tests verify owned transactions commit on success, roll back on failure, and close their sessions.
+- Composition tests verify decorated service calls borrow one caller-owned session without committing it.
-- Cache tests clear `get_session_factory` before and after replacing engines.
+- SAVEPOINT tests verify local rollback preserves the outer transaction and successful exit does not commit it.
-- Dependency override exists for the FastAPI session factory.
+- Tests that depend on SAVEPOINT timing account for `begin_nested()` flushing pending state on entry.
 - Rollback behavior is verified for failed write units.
 - Parallel-task tests verify no shared AsyncSession instances.
-- Lifespan tests confirm session factory is initialized and teardown-safe.
+- Lifecycle tests confirm schema initialization, factory availability, deterministic teardown, and expected cache behavior.
 

docs/skills/async-fastapi-sqlmodel/references/sqlmodel.md

@@ -12,7 +12,7 @@
     - Status: adopted
     - Decision level: mandatory
     - Applies to: api-runtime, workers, tests
-    - Last reviewed: 2026-07-26
+    - Last reviewed: 2026-08-06
 
 ---
 
@@ -34,7 +34,7 @@ SQLModel is designed for FastAPI, built on Pydantic and SQLAlchemy, and intended
 ## Rules
 
 - Default to SQLModel for new table models and API data models.
-- Keep SQLAlchemy async primitives as the runtime base: `create_async_engine`, `async_sessionmaker`, and `AsyncSession`.
+- Keep SQLAlchemy engine and factory primitives as the runtime base: `create_async_engine` and `async_sessionmaker`. For SQLModel applications, use SQLModel's `AsyncSession` wrapper so its typed `exec()` API remains available.
 - Keep transaction and session ownership policies identical whether models are SQLAlchemy Declarative or SQLModel.
 - Use SQLModel inheritance to share validated fields while keeping table, create, update, and public contracts distinct where their semantics differ.
 - Use SQLAlchemy declarative models only for a concrete unsupported mapping or third-party constraint; document the reason.
@@ -72,17 +72,14 @@ class UserRead(UserBase):
 ### Pattern B: Keep SQLModel models with the async runtime
 
 ```python
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlmodel import select
 
-engine = create_async_engine(settings.database_url, pool_pre_ping=True)
+async with database_scope(settings.database_url) as session_factory:
-session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    async with session_factory() as session:
-
+    users = (await session.exec(select(User))).all()
-async with session_factory() as session:
-    users = (await session.scalars(select(User))).all()
 ```
 
-`sqlmodel.select()` keeps SQLModel's typing-oriented statement construction, while `AsyncSession.scalars()` and the surrounding lifecycle come from SQLAlchemy.
+`database_scope()` enters the cached engine lifecycle, initializes registered SQLModel metadata by default, and yields the application session factory while SQLModel supplies the model and statement layer. `sqlmodel.select()` keeps SQLModel's typing-oriented statement construction, and SQLModel's `AsyncSession` adds typed `exec()` results while retaining SQLAlchemy's async lifecycle and transaction behavior. Import `AsyncSession` from `sqlmodel.ext.asyncio.session` when working with SQLModel models; use SQLAlchemy's `AsyncSession` only when the code intentionally has no SQLModel dependency.
 
 ---
 

docs/skills/async-fastapi-sqlmodel/references/testing.md

@@ -0,0 +1,172 @@
+# Testing Database Targets and Data
+
+Use the same engine and session primitives in production and tests. Tests select a different URL and, when transaction isolation is required, bind a test session factory to one test-owned connection and outer transaction. They do not replace repositories, services, or SQLAlchemy mechanics with mocks.
+
+## Decision Table
+
+| Test need | Database target | Isolation approach | What it proves |
+|---|---|---|---|
+| Fast, serial application tests | `sqlite+aiosqlite://` | Per-test engine or connection-bound session factory over an outer transaction | ORM mappings and ordinary application behavior |
+| Async code using multiple simultaneous sessions | Named SQLite shared-cache URL or temporary SQLite file | Per-test schema or cleanup strategy | Concurrent-session behavior without a database server |
+| PostgreSQL-specific behavior | Dedicated PostgreSQL test database | Per-test outer transaction and SAVEPOINT | SQL, constraints, types, locking, and migrations that SQLite cannot represent |
+
+SQLite is a useful fast target, not a drop-in PostgreSQL substitute. Keep a small PostgreSQL integration suite for PostgreSQL-specific queries, extensions, row locking, JSON semantics, collations, isolation, and migration validation.
+
+## Shared Construction Primitives
+
+Make the application factory accept a database URL or settings object. Production, workers, and ordinary integration tests enter the same [`database_scope()`](session.md#database-and-convenience-scopes). Tests enter the lower-level [`engine_scope()`](engine.md#owning-engine-scope) only when they need direct engine or connection ownership for schema setup, an outer transaction, or engine-specific assertions:
+
+```python
+from collections.abc import AsyncGenerator
+
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from .engine import engine_scope
+
+
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def test_engine(database_url: str) -> AsyncGenerator[AsyncEngine]:
+    async with engine_scope(database_url) as engine:
+        yield engine
+```
+
+Production passes its `postgresql+asyncpg://...` URL to `database_scope()`. A local SQLite run passes `sqlite+aiosqlite:///./app.db`. Tests pass a dedicated test URL to `database_scope()` or `engine_scope()` and receive schema initialization, deterministic disposal, and engine-cache cleanup when the context exits. Do not create an engine during module import: that makes it easy for tests to retain the production URL before an override is applied.
+
+Use migrations to provision an integration database when migrations are part of the release contract. `metadata.create_all()` is appropriate for focused ORM tests only when it accurately represents the schema under test. Import all table models before creating metadata; [SQLModel documents that model-registration order matters](https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/#import-table-models).
+
+## Transactional Async Fixture
+
+For tests that exercise code which commits, start an outer transaction on one test connection. Bind a test `SessionFactory` to that connection with `join_transaction_mode="create_savepoint"`. SQLAlchemy documents this as its test-suite pattern: sessions created by the factory resolve their commits through SAVEPOINTs while fixture teardown rolls back the outer transaction.
+
+```python
+from collections.abc import AsyncGenerator
+
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from .session import SessionFactory
+
+
+@pytest_asyncio.fixture(scope="function", loop_scope="session")
+async def session_factory(test_engine: AsyncEngine) -> AsyncGenerator[SessionFactory]:
+    async with test_engine.connect() as connection:
+        transaction = await connection.begin()
+        factory = async_sessionmaker(
+            bind=connection,
+            class_=AsyncSession,
+            expire_on_commit=False,
+            join_transaction_mode="create_savepoint",
+        )
+        try:
+            yield factory
+        finally:
+            await transaction.rollback()
+```
+
+Each factory call still creates a distinct `AsyncSession`, matching [session factory mechanics](session.md#session-factory-mechanics). The factory belongs to the fixture's engine and outer transaction and must not escape either scope.
+
+For service tests that pass a caller-owned session into decorated or undecorated service functions, derive that session from the same factory:
+
+```python
+@pytest_asyncio.fixture(scope="function", loop_scope="session")
+async def session(session_factory: SessionFactory) -> AsyncGenerator[AsyncSession]:
+    async with session_factory() as test_session:
+        await test_session.begin()
+        yield test_session
+```
+
+The explicit `begin()` gives test code one visible transaction from the start. Session closure rolls back unfinished work; the outer connection transaction remains the final isolation boundary even if application code commits its SAVEPOINT.
+
+For FastAPI request tests, override `_get_session_factory` so the production `SessionDep` retains its session-creation and cleanup behavior while receiving the test-bound factory. Always remove the override after the test because [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/) are stored in a mutable application-level dictionary.
+
+```python
+from collections.abc import Generator
+
+import pytest
+from fastapi import FastAPI
+
+from .fastapi import _get_session_factory
+from .session import SessionFactory
+
+
+@pytest.fixture
+def app_with_test_database(app: FastAPI, session_factory: SessionFactory) -> Generator[FastAPI]:
+    def get_test_session_factory() -> SessionFactory:
+        return session_factory
+
+    app.dependency_overrides[_get_session_factory] = get_test_session_factory
+    try:
+        yield app
+    finally:
+        app.dependency_overrides.pop(_get_session_factory, None)
+```
+
+Construct `app` with test settings before lifespan starts so startup cannot resolve the production URL. The override changes request session creation; it does not prevent lifespan from entering its configured `database_scope()`.
+
+The connection-bound factory is deliberately serial even though it creates distinct sessions: those sessions still share one connection and outer transaction. A test that verifies concurrently active sessions must use independent connections and a database target that supports them.
+
+## SQLite Targets
+
+### Serial in-memory tests
+
+Use `sqlite+aiosqlite://` for a fresh in-memory database when the test runs all database work serially. SQLAlchemy's `aiosqlite` dialect uses a single-connection `StaticPool` for this target, so all sessions share one SQLite transaction state. One session's rollback can discard another session's uncommitted work.
+
+`engine_scope()` imports the model package and creates the schema by default, then disposes the engine and clears cached resolution deterministically:
+
+```python
+from collections.abc import AsyncGenerator
+
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import AsyncEngine
+from .engine import engine_scope
+
+
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def test_engine() -> AsyncGenerator[AsyncEngine]:
+    async with engine_scope("sqlite+aiosqlite://") as engine:
+        yield engine
+```
+
+### Concurrent in-memory tests
+
+Do not use the default `:memory:` target for tests that have multiple active sessions or tasks. Use a named shared-cache database instead, with a name unique to the test process:
+
+```text
+sqlite+aiosqlite:///file:test-suite?mode=memory&cache=shared&uri=true
+```
+
+This lets connections share the same in-memory database while retaining independent transaction state. A temporary file URL such as `sqlite+aiosqlite:////tmp/test.db` is often simpler when test isolation or cleanup tooling already manages files.
+
+For both SQLite forms, enable and test the constraints your application depends on. SQLite foreign-key enforcement is disabled by default, and its transaction behavior has driver-specific differences. Keep PostgreSQL integration coverage for behavior that SQLite cannot faithfully model.
+
+## Test Data Practices
+
+- Build only the data a test needs, through named factory functions or pytest fixtures rather than a large global seed.
+- Give each fixture a domain meaning, such as `active_account`, `expired_subscription`, or `admin_user`; avoid opaque rows with unexplained defaults.
+- Set values relevant to the assertion explicitly, including timestamps, permissions, statuses, and unique identifiers. Use fixed clocks or injected clock values instead of the wall clock.
+- Construct object graphs through relationships, then `await session.flush()` before reading generated identifiers or passing foreign keys onward. `flush()` exercises database constraints without ending the test transaction.
+- Seed prerequisite data before creating a client request. Let the endpoint own the mutation being asserted; do not pre-insert the row that the endpoint is supposed to create.
+- Use `commit()` in fixture setup only when the test specifically needs to prove post-commit behavior. With the transactional fixture, this remains isolated through the outer rollback.
+- Keep shared reference data immutable and explicit. If it must be reused for performance, load it once into a dedicated test database and reset all mutable tables between tests; never depend on test order.
+- Include both valid and constraint-breaking graphs where a behavior depends on foreign keys, uniqueness, nullability, or cascading deletes. SQLite-only tests should not be the sole evidence for PostgreSQL constraints.
+
+## Completion Checks
+
+- A test run cannot reach the production URL; production credentials are absent from the test environment.
+- Production PostgreSQL, local SQLite, and in-memory SQLite all use `database_scope()` unless a test explicitly needs lower-level engine or connection ownership.
+- Every test or fixture scope owns its override, session factory, connection, transaction, and session cleanup; the session-scoped engine fixture owns disposal and cache cleanup.
+- Request tests override `_get_session_factory`, preserving production request-session creation and cleanup behavior.
+- Test data is deterministic, minimal, and expresses the scenario under test.
+- PostgreSQL integration tests cover every PostgreSQL-specific contract and run against migrations where migrations are shipped.
+
+## Sources
+
+- [SQLAlchemy: joining a session into an external transaction](https://docs.sqlalchemy.org/en/21/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites)
+- [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
+- [SQLAlchemy SQLite dialect and async in-memory pooling](https://docs.sqlalchemy.org/en/21/dialects/sqlite.html#using-a-memory-database-with-multiple-coroutines)
+- [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/)
+- [SQLModel testing with FastAPI](https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/)
+- [pytest-asyncio fixtures](https://pytest-asyncio.readthedocs.io/en/stable/how-to-guides/index.html)

docs/skills/nicegui/references/fastapi-uvicorn-startup.md

@@ -126,7 +126,7 @@ def get_settings() -> Settings:
 The argument-free [`functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache) provider is appropriate here because both the project entry point and Uvicorn's zero-argument factory need process-lifetime access. Each reload or worker process gets its own settings instance. Do not add override arguments to `get_settings()`; inject a `Settings` instance directly into `create_app()` in tests or alternate composition roots. See the [Pydantic settings implementation guide](../../pydantic-settings/SKILL.md) for source precedence, independent settings boundaries, cache clearing, and runtime reload guidance.
 
 ```python title="src/my_app/main.py"
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 import uvicorn
@@ -137,7 +137,7 @@ from my_app.config import Settings, get_settings
 
 
 @asynccontextmanager
-async def lifespan(app: FastAPI) -> AsyncIterator[None]:
+async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
     app.state.ready = True
     try:
         yield

docs/skills/pydantic-settings/SKILL.md

@@ -203,11 +203,125 @@ Use independent `BaseSettings` classes when the objects have genuinely independe
 
 Construct independent objects explicitly at the composition root and inject each dependency. Do not nest one `BaseSettings` class inside another merely to reuse its fields. Extract a shared `BaseModel` schema when models need common structure.
 
+### Alternative Database Backends
+
+When one application can run against one of several database backends, model the selected backend as a [discriminated union](https://docs.pydantic.dev/latest/concepts/unions/#discriminated-unions). Pydantic validates only the variant selected by `driver`, so required PostgreSQL values do not make a SQLite configuration fail, and vice versa.
+
+```python
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class SqliteSettings(BaseModel):
+  driver: Literal["sqlite"] = "sqlite"
+  path: str = "app.db"
+
+
+class PostgresSettings(BaseModel):
+  driver: Literal["postgres"] = "postgres"
+  host: str
+  port: int = 5432
+  database: str
+  user: str
+  password: SecretStr
+
+
+DatabaseSettings = Annotated[
+  SqliteSettings | PostgresSettings,
+  Field(discriminator="driver"),
+]
+
+
+class Settings(BaseSettings):
+  model_config = SettingsConfigDict(
+    env_prefix="APP_",
+    env_nested_delimiter="__",
+    env_file=".env",
+    extra="ignore",
+    frozen=True,
+  )
+
+  database: DatabaseSettings
+```
+
+Choose one configuration. A SQLite deployment requires no PostgreSQL variables:
+
+```dotenv
+APP_DATABASE__DRIVER=sqlite
+APP_DATABASE__PATH=./data/app.db
+```
+
+A PostgreSQL deployment requires only the PostgreSQL branch:
+
+```dotenv
+APP_DATABASE__DRIVER=postgres
+APP_DATABASE__HOST=db.internal
+APP_DATABASE__PORT=5432
+APP_DATABASE__DATABASE=app
+APP_DATABASE__USER=app_user
+APP_DATABASE__PASSWORD=provided-by-the-runtime
+```
+
+After settings validation, select an async SQLAlchemy driver URL. This is a pure configuration step; create the engine, session factory, and sessions in their own lifecycle-managed providers:
+
+```python
+from functools import cache
+
+from sqlalchemy import URL
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import create_async_engine
+
+
+def get_database_url(settings: Settings) -> str:
+    match settings.database:
+        case SqliteSettings(path=path):
+            url = URL.create(
+                drivername="sqlite+aiosqlite",
+                database=path,
+            )
+        case PostgresSettings() as database:
+            url = URL.create(
+                drivername="postgresql+asyncpg",
+                host=database.host,
+                port=database.port,
+                database=database.database,
+                username=database.user,
+                password=database.password.get_secret_value(),
+            )
+    return url.render_as_string(hide_password=False)
+
+
+@cache
+def get_engine(database_url: str) -> AsyncEngine:
+    return create_async_engine(database_url, pool_pre_ping=True)
+
+
+async def dispose_engine(database_url: str) -> None:
+  engine = get_engine(database_url)
+  try:
+    await engine.dispose()
+  finally:
+    get_engine.cache_clear()
+
+
+async def refresh_engine(database_url: str) -> AsyncEngine:
+    await dispose_engine(database_url)
+    return get_engine(database_url)
+```
+
+At the composition boundary, resolve the URL once with `get_database_url(settings)` and use it to retrieve the cached engine. In FastAPI, expose that engine through lifespan and build one `async_sessionmaker` from it; each request or unit of work then creates its own `AsyncSession`. Do not call `aiosqlite.connect()` or `asyncpg.create_pool()` directly: `aiosqlite` and `asyncpg` are selected as SQLAlchemy drivers by the URL, while SQLAlchemy owns pooling, disposal, and session integration.
+
+The nested variants remain `BaseModel` classes. `Settings` is the only `BaseSettings` model and therefore the only object that reads environment variables, dotenv files, or secrets. This keeps one source policy and validated configuration snapshot while keeping the engine, session factory, and sessions in their distinct lifecycles. See the [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html), the [engine lifecycle guidance](../async-fastapi-sqlmodel/references/engine.md), and the [session lifecycle guidance](../async-fastapi-sqlmodel/references/session.md).
+
 Quality gate:
 
 1. Nested sections share one source policy and lifecycle.
 2. Independent settings have distinct owners, prefixes, or lifecycles.
 3. The application does not repeatedly scan the same sources through accidental nested `BaseSettings` construction.
+4. Each backend configuration validates without values required only by another backend.
+5. One cached `AsyncEngine` exists per configured driver URL, while each request or unit of work receives a new `AsyncSession`.
 
 ### 7. Own The Settings Lifecycle
 

tests/web/conftest.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 import pytest
@@ -14,7 +14,7 @@ from personal_mcp.web.app import create_app
 
 
 @pytest_asyncio.fixture
-async def client() -> AsyncIterator[AsyncClient]:
+async def client() -> AsyncGenerator[AsyncClient]:
     """Provides an AsyncClient bound to a fresh application instance."""
     app = create_app()
     async with AsyncClient(
@@ -30,7 +30,7 @@ def mcp_session_factory():
     """Provides an in-process context manager factory for MCP SDK sessions."""
 
     @asynccontextmanager
-    async def create_session(*, initialize: bool = True) -> AsyncIterator[ClientSession]:
+    async def create_session(*, initialize: bool = True) -> AsyncGenerator[ClientSession]:
         app = create_app()
         mcp_url = f"http://testserver{app.state.settings.mounts.mcp}"
         async with (

zensical.toml

@@ -65,6 +65,8 @@ nav = [
   ] },
   { "Prompts" = [
     { "Authoring" = "prompts/authoring/PROMPT.md" },
+    { "JSFiddle Page Layout" = "prompts/jsfiddle-page-layout/PROMPT.md" },
+    { "NiceGUI Component Extraction" = "prompts/nicegui-component-extraction/PROMPT.md" },
     { "Pytest Fill Scaffold" = "prompts/pytest-fill-scaffold/PROMPT.md" },
     { "Pytest Scaffold" = "prompts/pytest-scaffold/PROMPT.md" },
     { "Greenfield Architecture" = "prompts/greenfield-architecture/PROMPT.md" },
@@ -92,9 +94,11 @@ nav = [
       { "Overview" = "skills/async-fastapi-sqlmodel/SKILL.md" },
       { "Engine" = "skills/async-fastapi-sqlmodel/references/engine.md" },
       { "Session" = "skills/async-fastapi-sqlmodel/references/session.md" },
+      { "FastAPI" = "skills/async-fastapi-sqlmodel/references/fastapi.md" },
       { "Tx" = "skills/async-fastapi-sqlmodel/references/transactions.md" },
       { "SQLModel" = "skills/async-fastapi-sqlmodel/references/sqlmodel.md" },
       { "CRUD" = "skills/async-fastapi-sqlmodel/references/crud.md" },
+      { "Testing" = "skills/async-fastapi-sqlmodel/references/testing.md" },
       { "IO" = "skills/async-fastapi-sqlmodel/references/implicit_io.md" },
       { "Obs" = "skills/async-fastapi-sqlmodel/references/observability.md" },
       { "Template" = "skills/async-fastapi-sqlmodel/references/template.md" },