Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
70695ff218 | ||
|
|
a238fb4dc3 | ||
|
|
b6f109cf91 |
@@ -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
|
||||||
|
|
||||||
@@ -133,7 +142,8 @@ from contextlib import AsyncExitStack, asynccontextmanager
|
|||||||
from collections.abc import AsyncIterator
|
from collections.abc import AsyncIterator
|
||||||
|
|
||||||
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) -> AsyncIterator[None]:
|
||||||
@@ -141,7 +151,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
|||||||
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
|
||||||
|
|
||||||
@@ -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:
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -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:
|
||||||
@@ -280,8 +281,8 @@ from collections.abc import AsyncIterator
|
|||||||
|
|
||||||
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]
|
||||||
@@ -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 AsyncIterator
|
||||||
|
|
||||||
|
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) -> AsyncIterator[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() -> AsyncIterator[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() -> AsyncIterator[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/)
|
||||||
@@ -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
|
||||||
|
|
||||||
|
|||||||
@@ -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" },
|
||||||
|
|||||||
Reference in New Issue
Block a user