8 Commits
Author SHA1 Message Date
John Lancaster cd11ea8255 fixed caching due to docs symlink 2026-07-31 15:39:09 -05:00
John Lancaster bc21643e8c jsfiddle prompt 2026-07-31 15:12:18 -05:00
John Lancaster 1ed5856db0 asyncgenerator fix 2026-07-30 20:51:16 -05:00
John Lancaster 70695ff218 toc update 2026-07-30 01:29:19 -05:00
John Lancaster a238fb4dc3 example of 2 database backends 2026-07-30 01:29:05 -05:00
John Lancaster b6f109cf91 fastapi updates 2026-07-30 01:28:39 -05:00
John Lancaster 3abafc4850 prune 2026-07-30 01:04:11 -05:00
John Lancaster d4c7952175 task tweak 2026-07-30 01:03:51 -05:00
16 changed files with 381 additions and 31 deletions
+2 -1
View File
@@ -52,7 +52,8 @@
"args": [ "args": [
"run", "run",
"uvicorn", "uvicorn",
"personal_mcp.main:app", "personal_mcp.main:create_app",
"--factory",
"--host", "--host",
"127.0.0.1", "127.0.0.1",
"--port", "--port",
+1 -1
View File
@@ -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=uv.lock,target=uv.lock \
--mount=type=bind,source=pyproject.toml,target=pyproject.toml \ --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
--mount=type=bind,source=src/,target=src/ \ --mount=type=bind,source=src/,target=src/ \
uv sync --no-editable uv sync --no-editable --refresh-package prompts
USER appuser USER appuser
-2
View File
@@ -6,5 +6,3 @@ services:
restart: unless-stopped restart: unless-stopped
ports: ports:
- "8765:8765" - "8765:8765"
volumes:
- ./docs:/app/src/personal_mcp/docs
@@ -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.0.0
tags:
- frontend
- html
- css
- jsfiddle
- layout
- prototyping
- prompts
capabilities:
- resource://prompts/jsfiddle-page-layout/document
arguments:
domain:
title: Domain
description: The product, service, organization, or subject the sample page should represent, including its audience when known.
required: true
layout_brief:
title: Layout brief
description: Optional page type, required sections, content priorities, visual direction, or constraints.
required: false
---
# 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.
## 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.
+20 -5
View File
@@ -110,6 +110,14 @@ Pool sizing, overflow, recycle, pre-ping, isolation, statement timeouts, and hea
See [observability and resilience](references/observability.md). 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 ## Reference Map
| Concept | Reference | | Concept | Reference |
@@ -123,6 +131,7 @@ See [observability and resilience](references/observability.md).
| Observability and resilience | [Observability reference](references/observability.md) | | Observability and resilience | [Observability reference](references/observability.md) |
| SQLModel-first modeling | [SQLModel integration reference](references/sqlmodel.md) | | SQLModel-first modeling | [SQLModel integration reference](references/sqlmodel.md) |
| CRUD repository and standalone functions | [Basic CRUD reference](references/crud.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 ## Canonical Composition Pattern
@@ -130,23 +139,28 @@ This example shows the ownership boundaries. Adapt state storage and dependency
```python ```python
from contextlib import AsyncExitStack, asynccontextmanager from contextlib import AsyncExitStack, asynccontextmanager
from collections.abc import AsyncIterator from collections.abc import AsyncGeneratorr
from fastapi import FastAPI from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]: async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
async with AsyncExitStack() as stack: async with AsyncExitStack() as stack:
engine = create_async_engine(settings.database_url) engine = create_async_engine(settings.database_url)
stack.push_async_callback(engine.dispose) stack.push_async_callback(engine.dispose)
session_factory = async_sessionmaker(engine, expire_on_commit=False) session_factory = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
)
app.state.session_factory = session_factory app.state.session_factory = session_factory
yield yield
async def get_session() -> AsyncIterator[AsyncSession]: async def get_session() -> AsyncGenerator[AsyncSession]:
async with app.state.session_factory() as session: async with app.state.session_factory() as session:
yield session yield session
``` ```
@@ -173,6 +187,7 @@ When reviewing code, verify:
- Relationship and deferred-column access cannot surprise the event loop with implicit I/O. - Relationship and deferred-column access cannot surprise the event loop with implicit I/O.
- Pool and timeout settings are justified by deployment behavior. - Pool and timeout settings are justified by deployment behavior.
- Tests exercise rollback, cleanup, concurrency, and lifespan behavior where relevant. - 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 ## Anti-Patterns to Flag
@@ -61,8 +61,8 @@ This reference uses direct field arguments and full-update semantics to keep the
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. 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`.
```python ```python
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from .session import session_scope from .session import session_scope
from .session import transaction_scope from .session import transaction_scope
@@ -170,7 +170,7 @@ Update and delete load the row through the same session that mutates it. This av
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 stores repeatable database configuration, never a mutable session. Every method delegates to the analogous function and exposes the same optional-session contract.
```python ```python
from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel.ext.asyncio.session import AsyncSession
class WidgetRepository: class WidgetRepository:
def __init__(self, database_url: str) -> None: def __init__(self, database_url: str) -> None:
@@ -64,14 +64,14 @@ Resolve settings at the composition boundary and call `get_engine(settings.datab
The lifespan context manager only connects the cached resource to FastAPI ownership: The lifespan context manager only connects the cached resource to FastAPI ownership:
```python ```python
from collections.abc import AsyncIterator from collections.abc import AsyncGeneratorr
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from fastapi import FastAPI from fastapi import FastAPI
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]: async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
database_url = app.state.settings.database_url database_url = app.state.settings.database_url
engine = get_engine(database_url) engine = get_engine(database_url)
app.state.engine = engine app.state.engine = engine
@@ -15,6 +15,7 @@ Purpose: concept registry for the principles, mechanics, and implementation guid
| Observability and resilience | [observability.md](observability.md) | adopted | mandatory | 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-07-26 |
| 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-07-26 |
| Test database targets and fixture data | [testing.md](testing.md) | adopted | mandatory | platform/backend | 2026-07-30 |
--- ---
@@ -81,7 +81,8 @@ Cache it by the application-owned engine so repeated composition calls return th
```python ```python
from functools import cache from functools import cache
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlmodel.ext.asyncio.session import AsyncSession
from .engine import dispose_engine from .engine import dispose_engine
from .engine import get_engine from .engine import get_engine
@@ -129,7 +130,7 @@ Otherwise, a later call can return a maker that still references the old engine
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: 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:
```python ```python
from collections.abc import AsyncIterator from collections.abc import AsyncGeneratorr
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
@@ -138,7 +139,7 @@ async def session_scope(
*, *,
database_url: str, database_url: str,
session: AsyncSession | None = None, session: AsyncSession | None = None,
) -> AsyncIterator[AsyncSession]: ) -> AsyncGenerator[AsyncSession]:
if session is not None: if session is not None:
yield session yield session
return return
@@ -171,7 +172,7 @@ async def transaction_scope(
*, *,
database_url: str, database_url: str,
session: AsyncSession | None = None, session: AsyncSession | None = None,
) -> AsyncIterator[AsyncSession]: ) -> AsyncGenerator[AsyncSession]:
if session is not None: if session is not None:
if not session.in_transaction(): if not session.in_transaction():
raise RuntimeError("A supplied session must have an active transaction") raise RuntimeError("A supplied session must have an active transaction")
@@ -226,7 +227,7 @@ Pass the database URL to repository constructors. The repository stores repeatab
```python ```python
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel.ext.asyncio.session import AsyncSession
async def find_item(session: AsyncSession, item_id: int) -> Item | None: async def find_item(session: AsyncSession, item_id: int) -> Item | None:
@@ -276,12 +277,12 @@ This preserves atomicity without making repository objects hold mutable `AsyncSe
## Canonical FastAPI Dependency Pattern ## Canonical FastAPI Dependency Pattern
```python ```python
from collections.abc import AsyncIterator from collections.abc import AsyncGenerator
from fastapi import Depends from fastapi import Depends
from fastapi import Request from fastapi import Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlmodel.ext.asyncio.session import AsyncSession
type SessionFactory = async_sessionmaker[AsyncSession] type SessionFactory = async_sessionmaker[AsyncSession]
@@ -293,7 +294,7 @@ def resolve_session_factory(request: Request) -> SessionFactory:
async def get_db_session( async def get_db_session(
session_factory: SessionFactory = Depends(resolve_session_factory), session_factory: SessionFactory = Depends(resolve_session_factory),
) -> AsyncIterator[AsyncSession]: ) -> AsyncGenerator[AsyncSession]:
async with session_factory() as session: async with session_factory() as session:
yield session yield session
``` ```
@@ -302,7 +303,7 @@ Route usage:
```python ```python
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel.ext.asyncio.session import AsyncSession
from .session import get_db_session from .session import get_db_session
@@ -34,7 +34,7 @@ SQLModel is designed for FastAPI, built on Pydantic and SQLAlchemy, and intended
## Rules ## Rules
- Default to SQLModel for new table models and API data models. - 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. - 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 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. - Use SQLAlchemy declarative models only for a concrete unsupported mapping or third-party constraint; document the reason.
@@ -72,8 +72,9 @@ class UserRead(UserBase):
### Pattern B: Keep SQLModel models with the async runtime ### Pattern B: Keep SQLModel models with the async runtime
```python ```python
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlmodel import select from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
engine = create_async_engine(settings.database_url, pool_pre_ping=True) engine = create_async_engine(settings.database_url, pool_pre_ping=True)
session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
@@ -82,7 +83,7 @@ async with session_factory() as session:
users = (await session.scalars(select(User))).all() 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. `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.
--- ---
@@ -0,0 +1,154 @@
# Testing Database Targets and Data
Use the same application database construction path in production and tests. Tests select a different URL and bind their request-session dependency to a test-scoped 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 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.
## One Construction Path
Make the application factory accept a database URL or settings object, and keep engine and session-factory construction in one function. The only test-specific inputs should be the URL and, for request tests, the session dependency override.
```python
from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession
def create_database(
database_url: str,
) -> tuple[AsyncEngine, async_sessionmaker[AsyncSession]]:
engine = create_async_engine(database_url)
session_factory = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
)
return engine, session_factory
```
Production passes its `postgresql+asyncpg://...` URL to `create_database()`. A local SQLite run passes `sqlite+aiosqlite:///./app.db`. Tests pass a dedicated test URL to the same function. Do not create an engine during module import: that makes it easy for tests to accidentally 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 calls `commit()`, start an outer transaction on one test connection. Bind the test `AsyncSession` to that connection and use `join_transaction_mode="create_savepoint"`. SQLAlchemy documents this as its test-suite pattern: session commits resolve a SAVEPOINT while fixture teardown rolls back the outer transaction.
```python
from collections.abc import AsyncGeneratorr
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlmodel.ext.asyncio.session import AsyncSession
@pytest_asyncio.fixture
async def session(test_engine: AsyncEngine) -> AsyncGenerator[AsyncSession]:
async with test_engine.connect() as connection:
transaction = await connection.begin()
test_session = AsyncSession(
bind=connection,
expire_on_commit=False,
join_transaction_mode="create_savepoint",
)
try:
yield test_session
finally:
await test_session.close()
await transaction.rollback()
```
Use the test session through the normal FastAPI dependency seam, and always remove the override after the test. [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/) are an application-level dictionary, so leaving one installed leaks test state.
```python
import pytest
from fastapi import FastAPI
from sqlmodel.ext.asyncio.session import AsyncSession
@pytest.fixture
def app_with_test_session(
app: FastAPI,
session: AsyncSession,
) -> FastAPI:
async def get_test_session() -> AsyncGenerator[AsyncSession]:
yield session
app.dependency_overrides[get_session] = get_test_session
try:
yield app
finally:
app.dependency_overrides.clear()
```
This fixture is deliberately serial: one mutable `AsyncSession` must not serve concurrent tasks. A test that verifies concurrently active sessions should create independent sessions from a factory and use a database target that supports independent connections.
## 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.
Create the schema and dispose the engine deterministically:
```python
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncEngine
@pytest_asyncio.fixture
async def test_engine() -> AsyncGenerator[AsyncEngine]:
engine, _ = create_database("sqlite+aiosqlite://")
async with engine.begin() as connection:
await connection.run_sync(SQLModel.metadata.create_all)
try:
yield engine
finally:
await engine.dispose()
```
### 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 the same engine/session-factory construction path.
- Every test owns its override, connection, transaction, session, and engine cleanup.
- 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/)
@@ -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. 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" ```python title="src/my_app/main.py"
from collections.abc import AsyncIterator from collections.abc import AsyncGeneratorr
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
import uvicorn import uvicorn
@@ -137,7 +137,7 @@ from my_app.config import Settings, get_settings
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]: async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
app.state.ready = True app.state.ready = True
try: try:
yield yield
+113
View File
@@ -203,11 +203,124 @@ 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. 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, 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,
user=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()
```
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: Quality gate:
1. Nested sections share one source policy and lifecycle. 1. Nested sections share one source policy and lifecycle.
2. Independent settings have distinct owners, prefixes, or lifecycles. 2. Independent settings have distinct owners, prefixes, or lifecycles.
3. The application does not repeatedly scan the same sources through accidental nested `BaseSettings` construction. 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 ### 7. Own The Settings Lifecycle
@@ -16,7 +16,6 @@ REQUIRED_LIBRARY_TAGS_BY_SKILL = {
"fastapi-uv-docker": {"fastapi", "uv", "uvicorn", "docker"}, "fastapi-uv-docker": {"fastapi", "uv", "uvicorn", "docker"},
"mcp-details": {"mcp", "fastmcp"}, "mcp-details": {"mcp", "fastmcp"},
"nicegui": {"nicegui", "fastapi"}, "nicegui": {"nicegui", "fastapi"},
"nicegui-ui-customization": {"nicegui", "fastapi"},
"pytesting": {"pytest", "testing", "fastapi", "asyncio", "anyio"}, "pytesting": {"pytest", "testing", "fastapi", "asyncio", "anyio"},
"python-logging": {"python", "logging"}, "python-logging": {"python", "logging"},
"python-typing": {"python", "typing"}, "python-typing": {"python", "typing"},
+3 -3
View File
@@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import AsyncIterator from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
import pytest import pytest
@@ -14,7 +14,7 @@ from personal_mcp.web.app import create_app
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def client() -> AsyncIterator[AsyncClient]: async def client() -> AsyncGenerator[AsyncClient]:
"""Provides an AsyncClient bound to a fresh application instance.""" """Provides an AsyncClient bound to a fresh application instance."""
app = create_app() app = create_app()
async with AsyncClient( async with AsyncClient(
@@ -30,7 +30,7 @@ def mcp_session_factory():
"""Provides an in-process context manager factory for MCP SDK sessions.""" """Provides an in-process context manager factory for MCP SDK sessions."""
@asynccontextmanager @asynccontextmanager
async def create_session(*, initialize: bool = True) -> AsyncIterator[ClientSession]: async def create_session(*, initialize: bool = True) -> AsyncGenerator[ClientSession]:
app = create_app() app = create_app()
mcp_url = f"http://testserver{app.state.settings.mounts.mcp}" mcp_url = f"http://testserver{app.state.settings.mounts.mcp}"
async with ( async with (
+1
View File
@@ -95,6 +95,7 @@ nav = [
{ "Tx" = "skills/async-fastapi-sqlmodel/references/transactions.md" }, { "Tx" = "skills/async-fastapi-sqlmodel/references/transactions.md" },
{ "SQLModel" = "skills/async-fastapi-sqlmodel/references/sqlmodel.md" }, { "SQLModel" = "skills/async-fastapi-sqlmodel/references/sqlmodel.md" },
{ "CRUD" = "skills/async-fastapi-sqlmodel/references/crud.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" }, { "IO" = "skills/async-fastapi-sqlmodel/references/implicit_io.md" },
{ "Obs" = "skills/async-fastapi-sqlmodel/references/observability.md" }, { "Obs" = "skills/async-fastapi-sqlmodel/references/observability.md" },
{ "Template" = "skills/async-fastapi-sqlmodel/references/template.md" }, { "Template" = "skills/async-fastapi-sqlmodel/references/template.md" },