john/prompts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

template updates

Browse Source
This commit is contained in:
John Lancaster
2026-08-06 23:17:50 -05:00
parent 0dc06f72ca
commit 7ac90d29dd
8 changed files with 352 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/skills/async-fastapi-sqlmodel/references/engine.md
+74 -34
View File
@@ -2,73 +2,106 @@
!!! 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)
- [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 for each application, worker, command, or test lifecycle.
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-operation object.
- The composition root owns engine creation and disposal.
- `get_engine(database_url)` owns URL-keyed engine construction and caching.
- 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 for each application-owned engine lifecycle.
- Exactly one cached engine for each database URL during an active application-owned lifecycle.
- Exactly one active owning `engine_scope()` for a given URL.
- Zero `create_async_engine(...)` calls in feature code.
- Zero engine lookup or disposal calls in repository code.
---
## One Engine Context Manager
## Cached Engine Resolution
Use one [`asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager) to pair engine creation with disposal:
[`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
from sqlalchemy.ext.asyncio import create_async_engine
@cache
def get_engine(database_url: str) -> AsyncEngine:
engine = create_async_engine(database_url, pool_pre_ping=True)
if engine.dialect.name == "sqlite":
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 sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession
type SessionFactory = async_sessionmaker[AsyncSession]
from sqlmodel import SQLModel
@asynccontextmanager
async def database_scope(database_url: str) -> AsyncGenerator[SessionFactory]:
async with engine_scope(database_url) as engine:
yield async_sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)
@asynccontextmanager
async def engine_scope(database_url: str) -> AsyncGenerator[AsyncEngine]:
engine = create_async_engine(database_url, pool_pre_ping=True)
if engine.dialect.name == "sqlite":
configure_aiosqlite_engine(engine)
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:
engine = get_engine(database_url)
try:
await engine.dispose()
finally:
get_engine.cache_clear()
```
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.
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.
Creating an `AsyncEngine` configures its dialect and pool; the first database operation normally establishes a connection. No cache is required when the application composition root enters this context exactly once. Removing the cache also removes cache-key, refresh, and invalidation behavior that otherwise must remain synchronized with the session factory.
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.
Resolve settings before entering the scope. Do not hide settings lookup or engine creation inside feature code.
`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.
Workers, scripts, and other composition roots enter `database_scope()` directly:
@@ -77,9 +110,11 @@ async with database_scope(settings.database_url) as session_factory:
await run_worker(session_factory)
```
For several fixed databases, nest one scope per engine. Use [`AsyncExitStack`](https://docs.python.org/3/library/contextlib.html#contextlib.AsyncExitStack) only when the number of engines is dynamic or conditional.
`database_scope()` is defined in [session management](session.md). It enters `engine_scope()` and creates the factory bound to the yielded engine.
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; no global cache reset is needed.
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 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.
See [FastAPI database integration](fastapi.md) for adapting `database_scope()` to application lifespan and dependency injection.
@@ -192,10 +227,11 @@ engine = create_async_engine(
## 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.
@@ -221,7 +257,9 @@ This prevents broken socket state and cross-process connection corruption.
- Resolve an engine from repositories instead of injecting a session dependency.
- Keep engine creation as a hidden side effect of import-time module globals.
- Keep a session factory alive after its bound engine scope exits.
- Add process-global engine caching when one composition root already owns the lifecycle.
- 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.
@@ -229,13 +267,15 @@ This prevents broken socket state and cross-process connection corruption.
## Engine Design Checklist
- One engine scope per application-owned database lifecycle.
- Engine creation and disposal paired by one framework-independent context manager.
- One cached engine per exact database URL during an active lifecycle.
- One owning engine scope per URL, with no overlapping owners.
- 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 feature-path engine creation.
- Tests enter the same scope and receive deterministic disposal without global cache cleanup.
- Tests enter the same scope and receive deterministic disposal plus cache cleanup.