. The middleware opened a session for the
+request, made it available as `db.session`, and closed it when the response was
+sent.
+
+!!! tip "That's the whole idea"
+ You never created or passed a session. `db.session` is bound to the current
+ request via a `ContextVar`, so it resolves correctly even from helper
+ functions called deep inside the request.
+
+## Configuring the engine
+
+Pass engine and session options through `engine_args` / `session_args`. These
+are forwarded verbatim to SQLAlchemy's
+[`create_async_engine`](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)
+and `async_sessionmaker`.
+
+```python
+app.add_middleware(
+ SQLAlchemyMiddleware,
+ db_url="postgresql+asyncpg://user:pass@localhost:5432/app",
+ engine_args={
+ "echo": True, # log every SQL statement
+ "pool_pre_ping": True, # validate connections before use
+ "pool_size": 5, # connections kept open
+ "max_overflow": 10, # extra connections allowed above pool_size
+ },
+ commit_on_exit=True, # commit the request session on a clean exit
+)
+```
+
+## Using the session outside a request
+
+Outside the request/response cycle (startup hooks, scripts, workers) there is no
+middleware to open a session for you. Open one explicitly with the `db` context
+manager:
+
+```python
+async def warm_cache():
+ async with db():
+ result = await db.session.execute(files.select())
+ return result.fetchall()
+
+
+@app.on_event("startup")
+async def on_startup():
+ await warm_cache()
+```
+
+## Next steps
+
+
+
+- :material-database-sync: [__Sessions & contexts__](guide/sessions.md) β how `db.session` and `async with db()` behave
+- :material-engine: [__Engine lifecycle__](guide/engine-lifecycle.md) β ownership, disposal, graceful shutdown
+- :material-arrow-decision: [__Concurrent queries__](guide/concurrency.md) β run parallel queries safely
+- :material-book-open-variant: [__API reference__](api-reference.md) β every public symbol
+
+
diff --git a/docs/guide/concurrency.md b/docs/guide/concurrency.md
new file mode 100644
index 0000000..4bf7c97
--- /dev/null
+++ b/docs/guide/concurrency.md
@@ -0,0 +1,127 @@
+# Concurrent Queries
+
+A single `AsyncSession` cannot run two operations at once β concurrent use
+raises SQLAlchemy's `InvalidRequestError: This session is provisioning a new
+connection; concurrent operations are not permitted`. To run queries in
+**parallel** you need a session **per task**, and you need to keep the number of
+simultaneous sessions under your connection-pool limit.
+
+`multi_sessions` mode solves both problems.
+
+## The problem with sharing one session
+
+```python
+# β Don't do this β all tasks share the one request session
+async def bad():
+ await asyncio.gather(
+ db.session.execute(text("SELECT 1")),
+ db.session.execute(text("SELECT 2")), # concurrent op on same session
+ )
+```
+
+## `multi_sessions=True`
+
+Opening `db(multi_sessions=True)` switches `db.session` to give **each task its
+own session**, tracked and cleaned up by the middleware:
+
+```python
+import asyncio
+from sqlalchemy import text
+
+async def run():
+ async with db(multi_sessions=True):
+ async def worker(n: int):
+ # each task gets a distinct session
+ return await db.session.execute(text(f"SELECT {n}"))
+
+ await asyncio.gather(*(worker(i) for i in range(5)))
+```
+
+Child task sessions are committed/rolled back and closed for you as each task
+finishes. All child tasks **must complete before the `async with` block exits**.
+
+## Throttling with `max_concurrent`
+
+Unbounded parallelism will exhaust the pool and raise
+`TimeoutError: QueuePool limit ... reached`. Set `max_concurrent` to cap the
+number of sessions holding a connection at once. When you do, child tasks must
+acquire their session through **`db.connection()`** or **`db.gather()`** so the
+middleware owns both the session lifetime and the semaphore slot.
+
+### `db.gather()` β the easy path
+
+A drop-in, pool-aware replacement for `asyncio.gather`. Each coroutine acquires
+a slot (and a session) before it runs and releases it afterwards:
+
+```python
+async def do_work(n: int) -> int:
+ async with db.connection() as session:
+ result = await session.execute(text(f"SELECT {n}"))
+ return result.scalar_one()
+
+
+async def run():
+ async with db(multi_sessions=True, max_concurrent=10):
+ results = await db.gather(*(do_work(i) for i in range(100)))
+ # never more than 10 connections in flight
+```
+
+!!! warning "Pass coroutines, not Tasks/Futures"
+ When `max_concurrent` is set, `db.gather()` accepts **coroutine objects
+ only**. A pre-created `Task` or `Future` may already be running outside the
+ semaphore, so it is rejected with `TypeError`. Pass `do_work(i)`, not
+ `asyncio.create_task(do_work(i))`.
+
+### `db.connection()` β explicit slots
+
+When you create your own tasks, open the session inside each task with
+`db.connection()`. It waits for a free slot before creating the session and
+releases it when the block exits:
+
+```python
+async def run():
+ async with db(multi_sessions=True, max_concurrent=10):
+ async def execute_query(query: str):
+ async with db.connection() as session:
+ return await session.execute(text(query))
+
+ tasks = [
+ asyncio.create_task(execute_query(f"SELECT {i}"))
+ for i in range(50)
+ ]
+ await asyncio.gather(*tasks)
+```
+
+Without `max_concurrent`, `db.connection()` still works β it just creates a
+session without throttling and cleans it up on exit.
+
+## Rules to remember
+
+- Child tasks that use the database **must finish before** the owning
+ `async with db(multi_sessions=True)` block exits. Tasks still parked on the
+ semaphore (or still running) when the block starts closing are cancelled.
+- With `max_concurrent` set, **direct `db.session` access from a child task is
+ rejected** β it isn't throttled. Use `db.connection()` or `db.gather()`
+ instead. (The parent task may still use `db.session` directly.)
+- Creating a new `db.connection()` session **after** the context has begun
+ closing raises `RuntimeError`.
+- `max_concurrent` must be `>= 1`, otherwise `ValueError` is raised.
+
+## Choosing an approach
+
+```mermaid
+flowchart TD
+ A[Need parallel DB work?] -->|no| B[Use db.session directly]
+ A -->|yes| C{Bounded by pool?}
+ C -->|"just a few tasks"| D["db(multi_sessions=True)"]
+ C -->|"many tasks"| E["db(multi_sessions=True, max_concurrent=N)"]
+ E --> F{Own your tasks?}
+ F -->|no, pass coroutines| G["db.gather(...)"]
+ F -->|yes, create_task| H["db.connection() inside each task"]
+```
+
+| Scenario | Use |
+| ------------------------------------------ | ---------------------------------------------- |
+| A handful of parallel queries | `db(multi_sessions=True)` + `db.session` |
+| Many queries, cap connections, simplest | `db(multi_sessions=True, max_concurrent=N)` + `db.gather()` |
+| Many queries, you manage the tasks | `db(multi_sessions=True, max_concurrent=N)` + `db.connection()` |
diff --git a/docs/guide/engine-lifecycle.md b/docs/guide/engine-lifecycle.md
new file mode 100644
index 0000000..c593105
--- /dev/null
+++ b/docs/guide/engine-lifecycle.md
@@ -0,0 +1,101 @@
+# Engine Lifecycle
+
+The middleware can either **own** the SQLAlchemy async engine or **borrow** one
+you created. Ownership decides who disposes the connection pool β and getting
+this wrong leaks connections on shutdown. This page makes the rules explicit.
+
+## Two ways to provide an engine
+
+=== "Middleware owns the engine (`db_url`)"
+
+ ```python
+ app.add_middleware(
+ SQLAlchemyMiddleware,
+ db_url="postgresql+asyncpg://user:pass@localhost/app",
+ engine_args={"pool_size": 5, "max_overflow": 10},
+ )
+ ```
+
+ The middleware calls `create_async_engine(db_url, **engine_args)`, **owns**
+ the result, and disposes it during ASGI lifespan shutdown.
+
+=== "You own the engine (`custom_engine`)"
+
+ ```python
+ from sqlalchemy.ext.asyncio import create_async_engine
+
+ engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/app")
+ app.add_middleware(SQLAlchemyMiddleware, custom_engine=engine)
+
+ # later, in your own shutdown / test cleanup:
+ await engine.dispose()
+ ```
+
+ The middleware uses your engine but **never disposes it**. Disposal is your
+ responsibility.
+
+!!! note "One of the two is required"
+ You must pass either `db_url` or `custom_engine`. Passing neither raises
+ `ValueError`.
+
+## Disposal during ASGI lifespan
+
+For a middleware-owned engine (`db_url`), disposal is automatic. It happens when
+the lifespan ends β including failure paths β so a raising shutdown handler
+can't leak the pool:
+
+- `lifespan.shutdown.complete`
+- `lifespan.shutdown.failed`
+- `lifespan.startup.failed`
+
+The engine is disposed **once**, for the application lifetime β not per request.
+
+```python
+from fastapi.testclient import TestClient
+
+# Running the lifespan (e.g. via TestClient context) triggers disposal:
+with TestClient(app):
+ ...
+# <- engine disposed here
+```
+
+!!! warning "Disposal blocks the shutdown ack"
+ Engine disposal runs **before** the lifespan acknowledgement is forwarded to
+ the ASGI server, so a slow pool drain delays graceful shutdown. Configure
+ your server's graceful-shutdown timeout (e.g. uvicorn's
+ `--timeout-graceful-shutdown`) to cover the worst-case time to close active
+ connections.
+
+## Manual disposal outside a lifespan
+
+When you build `SQLAlchemyMiddleware(db_url=...)` **outside** an ASGI lifespan β
+a script, an ad-hoc harness, a non-ASGI runtime β there is no
+`lifespan.shutdown` event, so nothing triggers disposal. Call `dispose()`
+yourself:
+
+```python
+middleware = SQLAlchemyMiddleware(app, db_url="postgresql+asyncpg://...")
+try:
+ ... # use db.session
+finally:
+ await middleware.dispose()
+```
+
+`dispose()` is:
+
+- **Idempotent on success** β calling it again is a no-op.
+- **Safe to retry on failure** β the proxy's session bindings are cleared
+ deterministically, so a later call actually re-attempts `engine.dispose()`
+ rather than silently no-op'ing on a half-disposed engine.
+- **A no-op for borrowed engines** β if you passed `custom_engine`, `dispose()`
+ does nothing; you own that engine.
+
+The same guidance applies to each pair returned by
+[`create_middleware_and_session_proxy()`](multi-database.md).
+
+## Summary
+
+| You pass | Engine created by | Disposed by | When |
+| ---------------- | ----------------- | ------------------------------- | -------------------------------------- |
+| `db_url` | the middleware | the middleware | lifespan shutdown, or `dispose()` |
+| `custom_engine` | you | you (`await engine.dispose()`) | whenever your own cleanup runs |
diff --git a/docs/guide/events.md b/docs/guide/events.md
new file mode 100644
index 0000000..70ba4bd
--- /dev/null
+++ b/docs/guide/events.md
@@ -0,0 +1,85 @@
+# SQLAlchemy Events
+
+SQLAlchemy's [event system](https://docs.sqlalchemy.org/en/20/orm/events.html)
+is **independent of the session and engine**. This middleware doesn't change how
+events fire β register listeners on your mapped classes (or on `Mapper` /
+`Session`) with `sqlalchemy.event.listens_for` exactly as you would in a
+synchronous SQLAlchemy setup.
+
+## Registering listeners
+
+```python
+from datetime import datetime
+from sqlalchemy import Column, DateTime, Integer, String, event
+from sqlalchemy.orm import DeclarativeBase
+
+
+class Base(DeclarativeBase):
+ pass
+
+
+class User(Base):
+ __tablename__ = "users"
+ id = Column(Integer, primary_key=True)
+ username = Column(String(50), unique=True, nullable=False)
+ created_at = Column(DateTime, default=datetime.utcnow)
+ updated_at = Column(DateTime, default=datetime.utcnow)
+
+
+@event.listens_for(User, "before_insert")
+def normalize(mapper, connection, target):
+ target.username = target.username.lower().strip()
+
+
+@event.listens_for(User, "before_update")
+def touch_updated_at(mapper, connection, target):
+ target.updated_at = datetime.utcnow()
+
+
+@event.listens_for(User, "after_insert")
+def log_insert(mapper, connection, target):
+ print(f"user created: id={target.id}")
+```
+
+These fire when the session flushes, just like always.
+
+## :warning: Mapper events are synchronous
+
+Mapper-level events receive a **synchronous** `connection` argument:
+
+`before_insert` Β· `after_insert` Β· `before_update` Β· `after_update` Β·
+`before_delete` Β· `after_delete`
+
+Inside these handlers:
+
+- **Do not** `await` anything.
+- **Do not** call async ORM APIs.
+
+```python
+@event.listens_for(User, "before_insert")
+def handler(mapper, connection, target):
+ # β
pure, synchronous work on `target`
+ target.username = target.username.strip().lower()
+ # β await db.session.execute(...) <- not allowed here
+```
+
+If you need async work after a write, do it **after** `await
+db.session.commit()` returns, or use `Session`-level events such as
+`after_flush` / `after_commit` and schedule the async work from there.
+
+## A complete example
+
+A runnable example with validation, automatic timestamps, audit logging, and a
+commented-out soft-delete hook lives in the repository at
+[`examples/events_example.py`](https://github.com/h0rn3t/fastapi-async-sqlalchemy/blob/main/examples/events_example.py).
+It wires the listeners above into a small FastAPI CRUD app:
+
+```python
+@app.post("/users")
+async def create_user(username: str, email: str, full_name: str | None = None):
+ async with db():
+ user = User(username=username, email=email, full_name=full_name)
+ db.session.add(user)
+ await db.session.commit() # before_insert / after_insert fire on flush
+ return {"id": user.id, "username": user.username}
+```
diff --git a/docs/guide/index.md b/docs/guide/index.md
new file mode 100644
index 0000000..a682eab
--- /dev/null
+++ b/docs/guide/index.md
@@ -0,0 +1,87 @@
+# User Guide
+
+The middleware is small but covers several distinct concerns. Pick the topic you
+need β each page is self-contained.
+
+
+
+- :material-database-sync:{ .lg .middle } __Sessions & Contexts__
+
+ ---
+
+ The `db.session` proxy, the `async with db()` context manager,
+ `commit_on_exit`, and when each session is created and closed.
+
+ [:octicons-arrow-right-24: Read](sessions.md)
+
+- :material-engine:{ .lg .middle } __Engine Lifecycle__
+
+ ---
+
+ Who owns the engine (`db_url` vs `custom_engine`), when it is disposed, and
+ how to dispose it manually outside an ASGI lifespan.
+
+ [:octicons-arrow-right-24: Read](engine-lifecycle.md)
+
+- :material-arrow-decision:{ .lg .middle } __Concurrent Queries__
+
+ ---
+
+ `multi_sessions`, `max_concurrent`, `db.gather()` and `db.connection()` β
+ parallel work without blowing past the connection pool.
+
+ [:octicons-arrow-right-24: Read](concurrency.md)
+
+- :material-download-network:{ .lg .middle } __Streaming Responses__
+
+ ---
+
+ Why streaming bodies need their own session and how to write one safely.
+
+ [:octicons-arrow-right-24: Read](streaming.md)
+
+- :material-transit-connection-variant:{ .lg .middle } __Multiple Databases__
+
+ ---
+
+ `create_middleware_and_session_proxy()` for independent apps or databases.
+
+ [:octicons-arrow-right-24: Read](multi-database.md)
+
+- :material-bell-ring:{ .lg .middle } __SQLAlchemy Events__
+
+ ---
+
+ Using `before_insert` / `after_update` and friends with async sessions.
+
+ [:octicons-arrow-right-24: Read](events.md)
+
+- :material-language-python:{ .lg .middle } __Type Hints__
+
+ ---
+
+ Annotate `db` with `DBSessionMeta` for full mypy / IDE support.
+
+ [:octicons-arrow-right-24: Read](type-hints.md)
+
+
+
+## Mental model
+
+```mermaid
+flowchart LR
+ A[HTTP request] --> B[SQLAlchemyMiddleware]
+ B -->|opens| C[AsyncSession bound to ContextVar]
+ C --> D[Route / service code
reads db.session]
+ D --> E{clean exit?}
+ E -->|yes + commit_on_exit| F[commit]
+ E -->|exception| G[rollback]
+ F --> H[close session]
+ G --> H
+ H --> I[response sent]
+```
+
+The session lives for the duration of the request context. Everything in the
+guide is a variation on this: opening extra contexts (`async with db()`),
+running many sessions at once (`multi_sessions`), or moving the lifetime into a
+streaming body.
diff --git a/docs/guide/multi-database.md b/docs/guide/multi-database.md
new file mode 100644
index 0000000..568b1f0
--- /dev/null
+++ b/docs/guide/multi-database.md
@@ -0,0 +1,102 @@
+# Multiple Databases
+
+The default `SQLAlchemyMiddleware` / `db` pair is bound to **one** engine. To
+talk to several independent databases, create a separate middleware/session
+proxy pair for each with `create_middleware_and_session_proxy()`.
+
+## Create one pair per database
+
+```python title="databases.py"
+from fastapi_async_sqlalchemy import create_middleware_and_session_proxy
+
+FirstSQLAlchemyMiddleware, first_db = create_middleware_and_session_proxy()
+SecondSQLAlchemyMiddleware, second_db = create_middleware_and_session_proxy()
+```
+
+Each call returns an independent `(middleware_class, db_proxy)` tuple with its
+own `ContextVar` state and engine binding.
+
+!!! info "The default pair is just a pre-made instance"
+ `SQLAlchemyMiddleware, db = create_middleware_and_session_proxy()` is exactly
+ how the package builds the defaults it exports. Calling it yourself gives you
+ additional, fully isolated pairs.
+
+## Wire them into the app
+
+```python title="main.py"
+from fastapi import FastAPI
+
+from databases import FirstSQLAlchemyMiddleware, SecondSQLAlchemyMiddleware
+from routes import router
+
+app = FastAPI()
+app.include_router(router)
+
+app.add_middleware(
+ FirstSQLAlchemyMiddleware,
+ db_url="postgresql+asyncpg://user:pass@localhost:5432/primary_db",
+ engine_args={"pool_size": 5, "max_overflow": 10},
+)
+app.add_middleware(
+ SecondSQLAlchemyMiddleware,
+ db_url="mysql+aiomysql://user:pass@localhost:3306/secondary_db",
+ engine_args={"pool_size": 5, "max_overflow": 10},
+)
+```
+
+## Use the right proxy in each route
+
+```python title="routes.py"
+from fastapi import APIRouter
+from sqlalchemy import column, table
+
+from databases import first_db, second_db
+
+router = APIRouter()
+files = table("ms_files", column("id"))
+
+
+@router.get("/first-db-files")
+async def get_files_from_first_db():
+ result = await first_db.session.execute(files.select())
+ return result.fetchall()
+
+
+@router.get("/second-db-files")
+async def get_files_from_second_db():
+ result = await second_db.session.execute(files.select())
+ return result.fetchall()
+```
+
+Each proxy resolves its own request-scoped session, so `first_db.session` and
+`second_db.session` never collide.
+
+## One proxy, one live engine
+
+A proxy is **bound to a single live engine**. Reusing the same proxy with a
+different live engine is rejected:
+
+```text
+RuntimeError: This SQLAlchemy session proxy is already bound to another live
+engine. Use create_middleware_and_session_proxy() for independent apps or
+databases.
+```
+
+This guard exists so requests can never silently switch to a different database
+binding. The fix is to use a **fresh pair** per app or database β exactly what
+`create_middleware_and_session_proxy()` is for.
+
+!!! tip "Rebinding after disposal"
+ The binding is cleared when the owning middleware's engine is disposed (via
+ lifespan shutdown or `await middleware.dispose()`). After disposal the proxy
+ is free to bind a new engine β useful in test suites that build and tear down
+ an app per test.
+
+## Concurrency still applies per proxy
+
+Each proxy supports the full [concurrency API](concurrency.md) independently:
+
+```python
+async with first_db(multi_sessions=True, max_concurrent=10):
+ results = await first_db.gather(*(work(i) for i in range(100)))
+```
diff --git a/docs/guide/sessions.md b/docs/guide/sessions.md
new file mode 100644
index 0000000..89fd84a
--- /dev/null
+++ b/docs/guide/sessions.md
@@ -0,0 +1,137 @@
+# Sessions & Contexts
+
+Everything in this library revolves around one idea: **`db.session` is an
+`AsyncSession` bound to the current async context.** This page explains how that
+session is created, where it lives, and how to open your own contexts.
+
+## The `db` proxy
+
+`db` is a global object exported from the package:
+
+```python
+from fastapi_async_sqlalchemy import db
+```
+
+It exposes a small surface:
+
+| Member | Kind | Purpose |
+| ------------------ | --------------------- | ---------------------------------------------------- |
+| `db.session` | property | The `AsyncSession` for the current context |
+| `db(...)` | callable | Open an explicit session context manager |
+| `db.connection()` | method | Throttled session context manager (multi-session) |
+| `db.gather(...)` | coroutine | Pool-aware `asyncio.gather` (multi-session) |
+
+`db.session` is backed by a [`ContextVar`][contextvar], so each request β each
+independent async context β sees its own session. You never pass it around.
+
+## Inside a request
+
+When `SQLAlchemyMiddleware` is installed, every HTTP request gets a session
+opened **before** your route runs and finalized **after** it returns:
+
+```python
+@app.get("/items/{item_id}")
+async def get_item(item_id: int):
+ item = await db.session.get(Item, item_id)
+ return item
+```
+
+The same session is visible from any function called during the request, no
+arguments required:
+
+```python
+async def load_item(item_id: int) -> Item | None:
+ # same session as the route β resolved from the request context
+ return await db.session.get(Item, item_id)
+
+
+@app.get("/items/{item_id}")
+async def get_item(item_id: int):
+ return await load_item(item_id)
+```
+
+### Commit on exit
+
+By default the request session is **not** committed for you β you call
+`await db.session.commit()` yourself. Set `commit_on_exit=True` to commit
+automatically when the request finishes cleanly:
+
+```python
+app.add_middleware(
+ SQLAlchemyMiddleware,
+ db_url="postgresql+asyncpg://user:pass@localhost/app",
+ commit_on_exit=True,
+)
+```
+
+The finalization rules are:
+
+- **Clean exit + `commit_on_exit=True`** β `commit()`, then `close()`.
+- **Clean exit + `commit_on_exit=False`** (default) β just `close()` (uncommitted
+ work is rolled back by closing).
+- **Exception** β `rollback()`, then `close()`. The original exception
+ propagates; a failure during rollback/commit/close is surfaced too.
+
+!!! warning "Commit/rollback errors are not swallowed"
+ If `commit()` fails, the middleware attempts a `rollback()` and raises, so a
+ write failure can never be reported to the client as success.
+
+## Outside a request: `async with db()`
+
+Anywhere there is no request context β startup/shutdown hooks, CLI scripts,
+background tasks, tests β open a session explicitly:
+
+```python
+async def get_db_fetch():
+ async with db():
+ result = await db.session.execute(foo.select())
+ return result.fetchall()
+```
+
+`db()` accepts the same finalization options as the middleware:
+
+```python
+async with db(commit_on_exit=True):
+ db.session.add(User(name="ada"))
+ # committed automatically on a clean exit
+```
+
+You can also pass `session_args` to override sessionmaker arguments for that one
+context:
+
+```python
+async with db(session_args={"expire_on_commit": True}):
+ ...
+```
+
+### `MissingSessionError`
+
+Accessing `db.session` with no active context raises
+[`MissingSessionError`](../api-reference.md#exceptions):
+
+```python
+# β no request, no `async with db()`
+result = await db.session.execute(foo.select()) # MissingSessionError
+```
+
+The fix is always the same β wrap the access in a context:
+
+```python
+async with db():
+ result = await db.session.execute(foo.select())
+```
+
+### `SessionNotInitialisedError`
+
+If you access `db.session` before any `SQLAlchemyMiddleware` has been
+constructed (so the sessionmaker doesn't exist yet), you get
+[`SessionNotInitialisedError`](../api-reference.md#exceptions) instead. Make sure
+`app.add_middleware(SQLAlchemyMiddleware, ...)` runs during app setup.
+
+## Where to go next
+
+- Run **many** sessions at once β [Concurrent Queries](concurrency.md)
+- Stream a large response body β [Streaming Responses](streaming.md)
+- Understand engine ownership and shutdown β [Engine Lifecycle](engine-lifecycle.md)
+
+[contextvar]: https://docs.python.org/3/library/contextvars.html
diff --git a/docs/guide/streaming.md b/docs/guide/streaming.md
new file mode 100644
index 0000000..adffb82
--- /dev/null
+++ b/docs/guide/streaming.md
@@ -0,0 +1,102 @@
+# Streaming Responses
+
+A `StreamingResponse` (or `FileResponse`) has a **different lifetime** from a
+normal request transaction. The body keeps yielding chunks *after* your route
+function returns, but the middleware-managed request session is tied to the
+request transaction β not to the stream. So you must not rely on `db.session`
+staying open while a streaming body runs.
+
+The rule: **open an explicit session inside the generator** so the body owns its
+own database lifetime.
+
+## The right way
+
+```python
+from fastapi.responses import StreamingResponse
+from fastapi_async_sqlalchemy import db
+
+@app.get("/export")
+async def export():
+ async def rows():
+ async with db(): # body-owned session
+ result = await db.session.stream(foo.select())
+ async for row in result:
+ yield f"{row.id}\n".encode()
+
+ return StreamingResponse(rows(), media_type="text/plain")
+```
+
+The `async with db()` inside the generator makes the session lifetime explicit
+and keeps the session open for the whole body.
+
+## Why not `commit_on_exit=True`?
+
+Implicit `commit_on_exit=True` is **not** a safe way to report streaming write
+success. The response may have already started β and early chunks already sent β
+before an unbounded body finishes. A late commit failure cannot un-send those
+chunks.
+
+To enforce this, the middleware actively rejects the unsafe combination. If a
+streaming response begins while `commit_on_exit=True` **and** the request
+session was already used, it raises:
+
+```text
+RuntimeError: `commit_on_exit=True` cannot use the middleware-managed request
+database session with a streaming response. Use `async with db()` inside the
+streaming generator, or manage the streaming transaction explicitly.
+```
+
+Similarly, once the request session has been closed for streaming, touching it
+again raises a `RuntimeError` telling you to use `async with db()` inside the
+generator.
+
+## If a streaming route needs to write
+
+Pick one of two explicit patterns:
+
+=== "Commit before streaming"
+
+ Complete and commit the write in its own context **before** creating the
+ streaming response, then stream read-only:
+
+ ```python
+ @app.post("/report")
+ async def make_report():
+ async with db(commit_on_exit=True):
+ db.session.add(ReportRun(status="started"))
+ # committed here, before any streaming begins
+
+ async def body():
+ async with db():
+ result = await db.session.stream(rows.select())
+ async for row in result:
+ yield serialize(row)
+
+ return StreamingResponse(body())
+ ```
+
+=== "Write inside the generator"
+
+ Make the generator own an explicit write transaction and design the API so
+ clients don't treat early chunks as confirmation of a completed write:
+
+ ```python
+ @app.get("/stream-and-write")
+ async def stream_and_write():
+ async def body():
+ async with db(commit_on_exit=True):
+ async for row in produce():
+ db.session.add(AuditRow(data=row))
+ yield row
+ # committed when the generator's context exits
+
+ return StreamingResponse(body())
+ ```
+
+## Migrating existing code
+
+If you previously used `db.session` directly inside a streaming generator, move
+that code into a generator-owned `async with db()` context as shown above. This
+keeps database access available for the whole body while making it clear that
+the session lifetime belongs to the stream, not the original request
+transaction.
diff --git a/docs/guide/type-hints.md b/docs/guide/type-hints.md
new file mode 100644
index 0000000..c4a6ac8
--- /dev/null
+++ b/docs/guide/type-hints.md
@@ -0,0 +1,88 @@
+# Type Hints
+
+The package ships a `py.typed` marker, so type checkers read its inline
+annotations. The one piece worth knowing about is how to annotate the `db`
+proxy itself.
+
+## Annotating `db` with `DBSessionMeta`
+
+Use `DBSessionMeta` when you need to type a
+function or attribute that holds the `db` proxy:
+
+```python
+from fastapi_async_sqlalchemy import DBSessionMeta, db
+
+
+def get_db() -> DBSessionMeta:
+ return db
+```
+
+This gives static checkers (mypy, pyright) and your IDE full autocomplete for
+the proxy surface β `session`, `connection()`, `gather()` and the `db(...)`
+call.
+
+## Runtime vs. type-check behavior
+
+`DBSessionMeta` is deliberately two things at once:
+
+- **At runtime** it is the actual metaclass of `db`, so identity and instance
+ checks work as they did in earlier versions:
+
+ ```python
+ from fastapi_async_sqlalchemy import DBSessionMeta, db
+
+ assert isinstance(db, DBSessionMeta)
+ assert type(db) is DBSessionMeta
+ ```
+
+- **At type-check time** it resolves to a structural
+ [`Protocol`](https://docs.python.org/3/library/typing.html#typing.Protocol)
+ describing the public API. That's what powers autocomplete and `mypy`
+ checking when you annotate with it.
+
+The Protocol surface is:
+
+```python
+class DBSessionMeta(Protocol):
+ @property
+ def session(self) -> AsyncSession: ...
+
+ def connection(self) -> AbstractAsyncContextManager[AsyncSession]: ...
+
+ async def gather(
+ self, *coros_or_futures: Any, return_exceptions: bool = ...
+ ) -> list[Any]: ...
+
+ def __call__(
+ self,
+ session_args: dict[str, Any] | None = ...,
+ commit_on_exit: bool = ...,
+ multi_sessions: bool = ...,
+ max_concurrent: int | None = ...,
+ ) -> AbstractAsyncContextManager[Any]: ...
+```
+
+## Dependency-injection style
+
+If you prefer passing `db` explicitly (e.g. for testability) rather than
+importing the global, the annotation makes it first-class:
+
+```python
+from fastapi import Depends
+from fastapi_async_sqlalchemy import DBSessionMeta, db
+
+
+def get_db() -> DBSessionMeta:
+ return db
+
+
+async def list_users(database: DBSessionMeta = Depends(get_db)):
+ result = await database.session.execute(users.select())
+ return result.fetchall()
+```
+
+## Works with SQLModel
+
+If `sqlmodel` is installed, the middleware uses `sqlmodel`'s `AsyncSession`
+subclass automatically, so `db.session` exposes SQLModel's session API. No extra
+configuration is needed.
diff --git a/docs/index.md b/docs/index.md
new file mode 100644
index 0000000..7fce099
--- /dev/null
+++ b/docs/index.md
@@ -0,0 +1,126 @@
+---
+hide:
+ - navigation
+ - toc
+---
+
+
+
+# FastAPI Async SQLAlchemy
+
+Drop-in async SQLAlchemy middleware for FastAPI. A request-scoped
+`AsyncSession` you reach through a single global `db` β no per-route
+dependency wiring, no manual session plumbing.
+
+
+[Get started :material-arrow-right:](getting-started.md){ .md-button .md-button--primary }
+[View on GitHub](https://github.com/h0rn3t/fastapi-async-sqlalchemy){ .md-button }
+
+
+
+[](https://pypi.org/project/fastapi-async-sqlalchemy/)
+[](https://pepy.tech/project/fastapi-async-sqlalchemy)
+[](https://opensource.org/licenses/MIT)
+[](https://github.com/h0rn3t/fastapi-async-sqlalchemy/actions)
+
+
+
+
+- :material-database-sync:{ .lg .middle } __Request-scoped sessions__
+
+ ---
+
+ A `ContextVar`-backed `AsyncSession` per request. Reach it from anywhere
+ with `db.session` β no dependency injection boilerplate.
+
+ [:octicons-arrow-right-24: Sessions & contexts](guide/sessions.md)
+
+- :material-engine:{ .lg .middle } __Engine lifecycle done right__
+
+ ---
+
+ Pass a `db_url` and the middleware owns and disposes the engine on
+ shutdown; pass a `custom_engine` and you keep ownership.
+
+ [:octicons-arrow-right-24: Engine lifecycle](guide/engine-lifecycle.md)
+
+- :material-arrow-decision:{ .lg .middle } __Pool-throttled concurrency__
+
+ ---
+
+ Run many queries in parallel without exhausting the pool. `db.gather()`
+ and `db.connection()` cap in-flight sessions at `max_concurrent`.
+
+ [:octicons-arrow-right-24: Concurrent queries](guide/concurrency.md)
+
+- :material-transit-connection-variant:{ .lg .middle } __Multiple databases__
+
+ ---
+
+ Build independent middleware/proxy pairs with
+ `create_middleware_and_session_proxy()` β one per database.
+
+ [:octicons-arrow-right-24: Multiple databases](guide/multi-database.md)
+
+- :material-download-network:{ .lg .middle } __Streaming-aware__
+
+ ---
+
+ Clear rules for `StreamingResponse` so the session lifetime belongs to the
+ body, not a closed request transaction.
+
+ [:octicons-arrow-right-24: Streaming responses](guide/streaming.md)
+
+- :material-language-python:{ .lg .middle } __Typed & SQLModel-ready__
+
+ ---
+
+ Ships `py.typed`, a `DBSessionMeta` Protocol for full autocomplete, and
+ works transparently with `sqlmodel`.
+
+ [:octicons-arrow-right-24: Type hints](guide/type-hints.md)
+
+
+
+## Installation
+
+```bash
+pip install fastapi-async-sqlalchemy
+```
+
+Requires Python 3.12+, `starlette>=0.40`, and `SQLAlchemy>=2.0`. Add the async
+driver for your database (`asyncpg`, `aiomysql`, `aiosqlite`, β¦).
+
+[contextvar]: https://docs.python.org/3/library/contextvars.html
diff --git a/examples/test_multisession_pool.py b/examples/test_multisession_pool.py
deleted file mode 100644
index 80f20c0..0000000
--- a/examples/test_multisession_pool.py
+++ /dev/null
@@ -1,57 +0,0 @@
-import asyncio
-
-import pytest
-from sqlalchemy import text
-from sqlalchemy.pool import AsyncAdaptedQueuePool
-
-from fastapi_async_sqlalchemy import create_middleware_and_session_proxy
-
-"""
-Goal: Ensure that session for each task is closed immediately after task completion
-to prevent session accumulation and connection pool exhaustion.
-"""
-
-# Create separate middleware for testing
-TestSQLAlchemyMiddleware, test_db = create_middleware_and_session_proxy()
-
-
-async def execute_query(query_id: int):
- """Execute query using session"""
- result = await test_db.session.execute(text(f"SELECT {query_id} as id"))
- # Simulate a long operation
- await asyncio.sleep(0.5) # 0.5-second delay
- return result.fetchone()
-
-
-@pytest.mark.asyncio
-async def test_multisession_with_limited_pool():
- """Test: 20 coroutines with multisession=True with a pool of 10 connections"""
-
- TestSQLAlchemyMiddleware(
- app=None,
- db_url="sqlite+aiosqlite:///test.db",
- engine_args={
- "poolclass": AsyncAdaptedQueuePool,
- "pool_size": 5,
- "max_overflow": 0,
- "echo": False,
- },
- )
-
- async with test_db(multi_sessions=True):
- # Create 20 coroutines
- tasks = [asyncio.create_task(execute_query(i)) for i in range(20)]
-
- # Execute all tasks in parallel
- results = await asyncio.gather(*tasks)
-
- # Checks
- assert len(results) == 20
- assert all(result is not None for result in results)
-
- print("β
Successfully executed 20 tasks")
- print(f"π Results: {[r[0] for r in results]}")
-
-
-if __name__ == "__main__":
- asyncio.run(test_multisession_with_limited_pool())
diff --git a/fastapi_async_sqlalchemy/__init__.py b/fastapi_async_sqlalchemy/__init__.py
index e6d95fa..94d8fd1 100644
--- a/fastapi_async_sqlalchemy/__init__.py
+++ b/fastapi_async_sqlalchemy/__init__.py
@@ -16,17 +16,15 @@
# created by ``create_middleware_and_session_proxy``) so ``isinstance(db,
# DBSessionMeta)`` and ``type(db) is DBSessionMeta`` keep working as in v0.5.
if TYPE_CHECKING:
- from fastapi_async_sqlalchemy._types import DBSessionMeta, DBSessionType
+ from fastapi_async_sqlalchemy._types import DBSessionMeta
else:
DBSessionMeta = type(db)
- DBSessionType = DBSessionMeta
__all__ = [
"db",
"SQLAlchemyMiddleware",
"create_middleware_and_session_proxy",
"DBSessionMeta",
- "DBSessionType",
]
__version__ = "0.8.0b1"
diff --git a/fastapi_async_sqlalchemy/_types.py b/fastapi_async_sqlalchemy/_types.py
index dba6a8f..b74ecce 100644
--- a/fastapi_async_sqlalchemy/_types.py
+++ b/fastapi_async_sqlalchemy/_types.py
@@ -58,6 +58,3 @@ def __call__(
) -> AbstractAsyncContextManager[Any]:
"""Open an explicit session context: ``async with db(): ...``."""
...
-
-
-DBSessionType = DBSessionMeta
diff --git a/mkdocs.yml b/mkdocs.yml
new file mode 100644
index 0000000..accc708
--- /dev/null
+++ b/mkdocs.yml
@@ -0,0 +1,124 @@
+site_name: FastAPI Async SQLAlchemy
+site_description: >-
+ Async SQLAlchemy middleware for FastAPI β a request-scoped AsyncSession
+ proxy with multi-database, pool-throttled concurrency, streaming and engine
+ lifecycle management.
+site_author: Eugene Shershen
+site_url: https://h0rn3t.github.io/fastapi-async-sqlalchemy/
+
+repo_name: h0rn3t/fastapi-async-sqlalchemy
+repo_url: https://github.com/h0rn3t/fastapi-async-sqlalchemy
+edit_uri: edit/main/docs/
+
+copyright: Copyright © Eugene Shershen β MIT License
+
+theme:
+ name: material
+ language: en
+ icon:
+ logo: material/database-sync
+ repo: fontawesome/brands/github
+ favicon: assets/favicon.svg
+ palette:
+ - media: "(prefers-color-scheme)"
+ toggle:
+ icon: material/brightness-auto
+ name: Switch to light mode
+ - media: "(prefers-color-scheme: light)"
+ scheme: default
+ primary: teal
+ accent: deep orange
+ toggle:
+ icon: material/weather-sunny
+ name: Switch to dark mode
+ - media: "(prefers-color-scheme: dark)"
+ scheme: slate
+ primary: teal
+ accent: amber
+ toggle:
+ icon: material/weather-night
+ name: Switch to system preference
+ font:
+ text: Inter
+ code: JetBrains Mono
+ features:
+ - navigation.tabs
+ - navigation.tabs.sticky
+ - navigation.sections
+ - navigation.top
+ - navigation.tracking
+ - navigation.indexes
+ - navigation.footer
+ - toc.follow
+ - search.suggest
+ - search.highlight
+ - search.share
+ - content.code.copy
+ - content.code.annotate
+ - content.tabs.link
+ - content.tooltips
+
+extra:
+ social:
+ - icon: fontawesome/brands/github
+ link: https://github.com/h0rn3t/fastapi-async-sqlalchemy
+ - icon: fontawesome/brands/python
+ link: https://pypi.org/project/fastapi-async-sqlalchemy/
+ generator: false
+
+extra_css:
+ - assets/extra.css
+
+plugins:
+ - search
+
+markdown_extensions:
+ - abbr
+ - admonition
+ - attr_list
+ - def_list
+ - footnotes
+ - md_in_html
+ - tables
+ - toc:
+ permalink: true
+ title: On this page
+ - pymdownx.betterem
+ - pymdownx.caret
+ - pymdownx.mark
+ - pymdownx.tilde
+ - pymdownx.keys
+ - pymdownx.details
+ - pymdownx.inlinehilite
+ - pymdownx.smartsymbols
+ - pymdownx.highlight:
+ anchor_linenums: true
+ line_spans: __span
+ pygments_lang_class: true
+ - pymdownx.superfences:
+ custom_fences:
+ - name: mermaid
+ class: mermaid
+ format: !!python/name:pymdownx.superfences.fence_code_format
+ - pymdownx.tabbed:
+ alternate_style: true
+ - pymdownx.tasklist:
+ custom_checkbox: true
+ - pymdownx.emoji:
+ emoji_index: !!python/name:material.extensions.emoji.twemoji
+ emoji_generator: !!python/name:material.extensions.emoji.to_svg
+
+nav:
+ - Home: index.md
+ - Getting Started: getting-started.md
+ - User Guide:
+ - guide/index.md
+ - Sessions & Contexts: guide/sessions.md
+ - Engine Lifecycle: guide/engine-lifecycle.md
+ - Concurrent Queries: guide/concurrency.md
+ - Streaming Responses: guide/streaming.md
+ - Multiple Databases: guide/multi-database.md
+ - SQLAlchemy Events: guide/events.md
+ - Type Hints: guide/type-hints.md
+ - API Reference: api-reference.md
+ - FAQ & Troubleshooting: faq.md
diff --git a/pyproject.toml b/pyproject.toml
index 51f1169..2e9080a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,3 +1,46 @@
+[build-system]
+requires = ["setuptools>=77"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "fastapi-async-sqlalchemy"
+dynamic = ["version"]
+description = "SQLAlchemy middleware for FastAPI"
+readme = "README.md"
+license = "MIT"
+authors = [{ name = "Eugene Shershen", email = "h0rn3t.null@gmail.com" }]
+requires-python = ">=3.12"
+dependencies = ["starlette>=0.40", "SQLAlchemy>=2.0"]
+classifiers = [
+ "Development Status :: 5 - Production/Stable",
+ "Environment :: Web Environment",
+ "Framework :: AsyncIO",
+ "Intended Audience :: Developers",
+ "Operating System :: OS Independent",
+ "Programming Language :: Python :: 3.12",
+ "Programming Language :: Python :: 3.13",
+ "Programming Language :: Python :: 3.14",
+ "Programming Language :: Python :: 3.15",
+ "Programming Language :: Python :: 3 :: Only",
+ "Programming Language :: Python :: Implementation :: CPython",
+ "Topic :: Internet :: WWW/HTTP :: HTTP Servers",
+ "Topic :: Internet :: WWW/HTTP :: Dynamic Content",
+ "Topic :: Software Development :: Libraries :: Python Modules",
+]
+
+[project.urls]
+Code = "https://github.com/h0rn3t/fastapi-async-sqlalchemy"
+"Issue tracker" = "https://github.com/h0rn3t/fastapi-async-sqlalchemy/issues"
+
+[tool.setuptools]
+packages = ["fastapi_async_sqlalchemy"]
+
+[tool.setuptools.package-data]
+fastapi_async_sqlalchemy = ["py.typed"]
+
+[tool.setuptools.dynamic]
+version = { attr = "fastapi_async_sqlalchemy.__version__" }
+
[tool.ruff]
line-length = 100
target-version = "py312"
diff --git a/requirements-docs.txt b/requirements-docs.txt
new file mode 100644
index 0000000..19058a1
--- /dev/null
+++ b/requirements-docs.txt
@@ -0,0 +1,2 @@
+# Documentation toolchain β install with: pip install -r requirements-docs.txt
+mkdocs-material>=9.5,<10
diff --git a/requirements.txt b/requirements.txt
index e514fb1..1604c53 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,42 +1,14 @@
-appdirs==1.4.3
-atomicwrites==1.3.0
-attrs>=19.3.0
-certifi>=2023.07.22
-chardet==3.0.4
-click>=8.1.3
-coverage>=5.2.1
-entrypoints==0.3
+# Runtime
+SQLAlchemy>=2.0
+starlette>=0.40
+
+# Test
fastapi>=0.115
-flake8==3.7.9
-idna==3.7
-importlib-metadata==1.5.0
-isort==5.13.2
-mccabe==0.6.1
-more-itertools==7.2.0
-packaging>=22.0
-pathspec>=0.9.0
-pluggy>=1.5.0
-pycodestyle==2.5.0
-pydantic>=2.7
-pyflakes==2.1.1
-pyparsing==2.4.2
+httpx>=0.20.0,<0.28.0
pytest>=8.3.0
+pytest-asyncio>=0.24.0
pytest-cov>=5.0.0
-PyYAML>=5.4
-regex>=2020.2.20
-requests>=2.22.0
-httpx>=0.20.0,<0.28.0
-six==1.12.0
-SQLAlchemy>=2.0
sqlmodel>=0.0.24
-asyncpg>=0.27.0
aiosqlite==0.20.0
-sqlparse>=0.5.4
-starlette>=0.40
-toml>=0.10.1
-urllib3>=1.25.9
-wcwidth==0.1.7
-zipp==3.19.1
-black==26.3.1
-pytest-asyncio>=0.24.0
+asyncpg>=0.27.0
greenlet>=3.2.4
diff --git a/setup.py b/setup.py
deleted file mode 100644
index 894b18b..0000000
--- a/setup.py
+++ /dev/null
@@ -1,48 +0,0 @@
-import re
-from pathlib import Path
-
-from setuptools import setup
-
-with open(Path("fastapi_async_sqlalchemy") / "__init__.py", encoding="utf-8") as fh:
- version = re.search(r'__version__ = "(.*?)"', fh.read(), re.M).group(1) # type: ignore
-
-with open("README.md", encoding="utf-8") as fh:
- long_description = fh.read()
-
-setup(
- name="fastapi-async-sqlalchemy",
- version=version,
- url="https://github.com/h0rn3t/fastapi-async-sqlalchemy.git",
- project_urls={
- "Code": "https://github.com/h0rn3t/fastapi-async-sqlalchemy",
- "Issue tracker": "https://github.com/h0rn3t/fastapi-async-sqlalchemy/issues",
- },
- license="MIT",
- author="Eugene Shershen",
- author_email="h0rn3t.null@gmail.com",
- description="SQLAlchemy middleware for FastAPI",
- long_description=long_description,
- long_description_content_type="text/markdown",
- packages=["fastapi_async_sqlalchemy"],
- package_data={"fastapi_async_sqlalchemy": ["py.typed"]},
- zip_safe=False,
- python_requires=">=3.12",
- install_requires=["starlette>=0.40", "SQLAlchemy>=2.0"],
- classifiers=[
- "Development Status :: 5 - Production/Stable",
- "Environment :: Web Environment",
- "Framework :: AsyncIO",
- "Intended Audience :: Developers",
- "License :: OSI Approved :: MIT License",
- "Operating System :: OS Independent",
- "Programming Language :: Python :: 3.12",
- "Programming Language :: Python :: 3.13",
- "Programming Language :: Python :: 3.14",
- "Programming Language :: Python :: 3.15",
- "Programming Language :: Python :: 3 :: Only",
- "Programming Language :: Python :: Implementation :: CPython",
- "Topic :: Internet :: WWW/HTTP :: HTTP Servers",
- "Topic :: Internet :: WWW/HTTP :: Dynamic Content",
- "Topic :: Software Development :: Libraries :: Python Modules",
- ],
-)
diff --git a/tests/test_additional_coverage.py b/tests/test_additional_coverage.py
deleted file mode 100644
index d2c63ff..0000000
--- a/tests/test_additional_coverage.py
+++ /dev/null
@@ -1,116 +0,0 @@
-"""
-Additional tests to reach target coverage of 97.22%
-"""
-
-import asyncio
-
-from fastapi import FastAPI
-
-
-def test_commit_on_exit_parameter():
- """Test commit_on_exit parameter in middleware initialization"""
- from sqlalchemy.ext.asyncio import create_async_engine
-
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- SQLAlchemyMiddleware, db = create_middleware_and_session_proxy()
- app = FastAPI()
-
- # Test commit_on_exit=True
- custom_engine = create_async_engine("sqlite+aiosqlite://")
- try:
- middleware = SQLAlchemyMiddleware(app, custom_engine=custom_engine, commit_on_exit=True)
- assert middleware.commit_on_exit is True
-
- # Test commit_on_exit=False (default)
- middleware2 = SQLAlchemyMiddleware(app, custom_engine=custom_engine, commit_on_exit=False)
- assert middleware2.commit_on_exit is False
- finally:
- asyncio.run(custom_engine.dispose())
-
-
-def test_exception_classes_simple():
- """Test exception classes are properly defined"""
- from fastapi_async_sqlalchemy.exceptions import MissingSessionError, SessionNotInitialisedError
-
- # Test exception instantiation without parameters
- missing_error = MissingSessionError()
- assert isinstance(missing_error, Exception)
-
- init_error = SessionNotInitialisedError()
- assert isinstance(init_error, Exception)
-
-
-def test_middleware_properties():
- """Test middleware properties and methods"""
- from fastapi import FastAPI
- from sqlalchemy.ext.asyncio import create_async_engine
-
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- SQLAlchemyMiddleware, db = create_middleware_and_session_proxy()
- app = FastAPI()
-
- # Test middleware properties
- custom_engine = create_async_engine("sqlite+aiosqlite://")
- try:
- middleware = SQLAlchemyMiddleware(app, custom_engine=custom_engine, commit_on_exit=True)
-
- assert hasattr(middleware, "commit_on_exit")
- assert middleware.commit_on_exit is True
- finally:
- asyncio.run(custom_engine.dispose())
-
-
-def test_basic_imports():
- """Test basic imports and module structure"""
- # Test main module imports
- from fastapi_async_sqlalchemy import SQLAlchemyMiddleware, db
-
- assert SQLAlchemyMiddleware is not None
- assert db is not None
-
- # Test exception imports
- from fastapi_async_sqlalchemy.exceptions import MissingSessionError, SessionNotInitialisedError
-
- assert MissingSessionError is not None
- assert SessionNotInitialisedError is not None
-
- # Test middleware module imports
- from fastapi_async_sqlalchemy.middleware import (
- DefaultAsyncSession,
- create_middleware_and_session_proxy,
- )
-
- assert create_middleware_and_session_proxy is not None
- assert DefaultAsyncSession is not None
-
-
-def test_middleware_factory_different_instances():
- """Test creating multiple middleware/db instances"""
- from fastapi import FastAPI
- from sqlalchemy.ext.asyncio import create_async_engine
-
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- # Create first instance
- SQLAlchemyMiddleware1, db1 = create_middleware_and_session_proxy()
-
- # Create second instance
- SQLAlchemyMiddleware2, db2 = create_middleware_and_session_proxy()
-
- # They should be different instances
- assert SQLAlchemyMiddleware1 is not SQLAlchemyMiddleware2
- assert db1 is not db2
-
- # Test both instances work
- app = FastAPI()
- engine = create_async_engine("sqlite+aiosqlite://")
-
- try:
- middleware1 = SQLAlchemyMiddleware1(app, custom_engine=engine)
- middleware2 = SQLAlchemyMiddleware2(app, custom_engine=engine)
-
- assert middleware1 is not middleware2
- finally:
- asyncio.run(engine.dispose())
diff --git a/tests/test_concurrent_queries_postgres.py b/tests/test_concurrent_queries_postgres.py
new file mode 100644
index 0000000..99abb76
--- /dev/null
+++ b/tests/test_concurrent_queries_postgres.py
@@ -0,0 +1,110 @@
+"""Real PostgreSQL/asyncpg reproduction of the `isce` concurrent-operations error.
+
+The production traceback originates here::
+
+ sqlalchemy.exc.InvalidRequestError: This session is provisioning a new
+ connection; concurrent operations are not permitted
+ (https://sqlalche.me/e/20/isce)
+
+The error is triggered by ``await asyncio.gather(session.execute(...),
+session.execute(...))`` against a single AsyncSession backed by asyncpg.
+SQLite/aiosqlite serialises internally and does NOT reliably reproduce it
+(see ``test_concurrent_queries.py``), so this module targets a real Postgres
+instance and is skipped unless ``POSTGRES_TEST_URL`` is set.
+
+Run locally::
+
+ POSTGRES_TEST_URL="postgresql+asyncpg://user:pass@localhost:5432/test" \\
+ pytest tests/test_concurrent_queries_postgres.py -v
+"""
+
+import asyncio
+import os
+import uuid
+
+import pytest
+from sqlalchemy import text
+
+POSTGRES_URL = os.getenv("POSTGRES_TEST_URL")
+
+pytestmark = pytest.mark.skipif(
+ not POSTGRES_URL,
+ reason="POSTGRES_TEST_URL not set; this test requires a real PostgreSQL/asyncpg instance",
+)
+
+
+@pytest.fixture
+def table_name():
+ return f"isce_repro_{uuid.uuid4().hex[:8]}"
+
+
+@pytest.fixture
+async def setup_table(app, db, SQLAlchemyMiddleware, table_name):
+ SQLAlchemyMiddleware(app, db_url=POSTGRES_URL)
+ async with db(commit_on_exit=True):
+ await db.session.execute(
+ text(f"CREATE TABLE {table_name} (id SERIAL PRIMARY KEY, name TEXT NOT NULL)")
+ )
+ await db.session.execute(
+ text(
+ f"INSERT INTO {table_name} (name) SELECT 'row_' || g FROM generate_series(1, 50) g"
+ )
+ )
+ try:
+ yield
+ finally:
+ async with db(commit_on_exit=True):
+ await db.session.execute(text(f"DROP TABLE IF EXISTS {table_name}"))
+
+
+@pytest.mark.asyncio
+async def test_gather_on_same_session_raises_isce(db, table_name, setup_table):
+ """asyncio.gather() on a single AsyncSession must raise isce on asyncpg."""
+ count_stmt = text(f"SELECT COUNT(*) FROM {table_name}")
+ rows_stmt = text(f"SELECT id, name FROM {table_name} LIMIT 10")
+
+ with pytest.raises(Exception) as exc_info:
+ async with db():
+ await asyncio.gather(
+ db.session.execute(count_stmt),
+ db.session.execute(rows_stmt),
+ )
+
+ error_msg = str(exc_info.value).lower()
+ assert (
+ "concurrent operations are not permitted" in error_msg
+ or "provisioning a new connection" in error_msg
+ or "isce" in error_msg
+ ), f"Expected isce error, got: {exc_info.value!r}"
+
+
+@pytest.mark.asyncio
+async def test_db_gather_multi_sessions_avoids_isce(db, table_name, setup_table):
+ """db.gather() in multi_sessions mode gives each task its own session β no isce."""
+ count_stmt = text(f"SELECT COUNT(*) FROM {table_name}")
+ rows_stmt = text(f"SELECT id, name FROM {table_name} LIMIT 10")
+
+ async def get_count():
+ result = await db.session.execute(count_stmt)
+ return result.scalar()
+
+ async def get_rows():
+ result = await db.session.execute(rows_stmt)
+ return result.fetchall()
+
+ async with db(multi_sessions=True, max_concurrent=2):
+ count, rows = await db.gather(get_count(), get_rows())
+
+ assert count == 50
+ assert len(rows) == 10
+
+
+@pytest.mark.asyncio
+async def test_sequential_execute_on_same_session_works(db, table_name, setup_table):
+ """Sequential awaits on the same session never trigger isce."""
+ async with db():
+ count_result = await db.session.execute(text(f"SELECT COUNT(*) FROM {table_name}"))
+ rows_result = await db.session.execute(text(f"SELECT id FROM {table_name} LIMIT 5"))
+
+ assert count_result.scalar() == 50
+ assert len(rows_result.fetchall()) == 5
diff --git a/tests/test_coverage_boost.py b/tests/test_coverage_boost.py
deleted file mode 100644
index 8d6baea..0000000
--- a/tests/test_coverage_boost.py
+++ /dev/null
@@ -1,132 +0,0 @@
-"""
-Simple tests to boost coverage to target level
-"""
-
-import asyncio
-from unittest.mock import AsyncMock
-
-import pytest
-from fastapi import FastAPI
-from sqlalchemy.exc import SQLAlchemyError
-
-
-def test_session_not_initialised_error():
- """Test SessionNotInitialisedError when accessing session without middleware"""
- from fastapi_async_sqlalchemy.exceptions import SessionNotInitialisedError
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- # Create fresh middleware/db instances - no middleware initialization
- SQLAlchemyMiddleware, db = create_middleware_and_session_proxy()
-
- # Should raise SessionNotInitialisedError (not MissingSessionError) when _Session is None
- with pytest.raises(SessionNotInitialisedError):
- _ = db.session
-
-
-def test_missing_session_error():
- """Test MissingSessionError when session context is None"""
- from fastapi_async_sqlalchemy.exceptions import MissingSessionError
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- SQLAlchemyMiddleware, db = create_middleware_and_session_proxy()
- app = FastAPI()
- SQLAlchemyMiddleware(app, db_url="sqlite+aiosqlite://")
-
- # Now _Session is initialized, but no active session context
- # This should raise MissingSessionError
- with pytest.raises(MissingSessionError):
- _ = db.session
-
-
-@pytest.mark.asyncio
-async def test_rollback_on_commit_exception():
- """Test rollback is called when commit raises exception (lines 114-116)"""
- # Create mock session that fails on commit
- mock_session = AsyncMock()
- mock_session.commit.side_effect = SQLAlchemyError("Commit failed!")
-
- # Create a simulated cleanup scenario
- async def test_cleanup():
- # This simulates the cleanup function with commit_on_exit=True
- try:
- await mock_session.commit()
- except Exception:
- await mock_session.rollback()
- raise
- finally:
- await mock_session.close()
-
- # Test that rollback is called when commit fails
- with pytest.raises(SQLAlchemyError):
- await test_cleanup()
-
- mock_session.rollback.assert_called_once()
- mock_session.close.assert_called_once()
-
-
-def test_import_fallbacks_work():
- """Test that import fallbacks are properly configured"""
- # Test async_sessionmaker import (lines 16-19)
- try:
- from sqlalchemy.ext.asyncio import async_sessionmaker
-
- # If available, use it
- assert async_sessionmaker is not None
- except ImportError: # pragma: no cover
- # Lines 18-19 would execute if async_sessionmaker not available
- from sqlalchemy.orm import sessionmaker as async_sessionmaker
-
- assert async_sessionmaker is not None
-
- # Test DefaultAsyncSession import (lines 22-27)
- from sqlalchemy.ext.asyncio import AsyncSession
-
- from fastapi_async_sqlalchemy.middleware import DefaultAsyncSession
-
- # Should be either SQLModel's AsyncSession or regular AsyncSession
- assert issubclass(DefaultAsyncSession, AsyncSession)
-
-
-def test_db_url_validation_with_none():
- """Test ValueError when db_url is explicitly None (line 58)"""
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- SQLAlchemyMiddleware, db = create_middleware_and_session_proxy()
- app = FastAPI()
-
- # Force the condition on line 58: db_url is None when custom_engine is not provided
- with pytest.raises(ValueError, match="You need to pass a db_url or a custom_engine parameter"):
- # This hits line 55 first, but let's also test a more specific case
- SQLAlchemyMiddleware(app, db_url=None, custom_engine=None)
-
-
-# Skipping the problematic test for now
-
-
-def test_skipped_tests_make_coverage():
- """Extra assertions to boost coverage a bit"""
- # Test basic imports work
- from fastapi_async_sqlalchemy import SQLAlchemyMiddleware, db
-
- assert SQLAlchemyMiddleware is not None
- assert db is not None
-
- from fastapi_async_sqlalchemy.exceptions import MissingSessionError, SessionNotInitialisedError
-
- assert MissingSessionError is not None
- assert SessionNotInitialisedError is not None
-
- # Test middleware with custom engine path
- from sqlalchemy.ext.asyncio import create_async_engine
-
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- SQLAlchemyMiddleware, db_fresh = create_middleware_and_session_proxy()
- app = FastAPI()
-
- custom_engine = create_async_engine("sqlite+aiosqlite://")
- try:
- middleware = SQLAlchemyMiddleware(app, custom_engine=custom_engine)
- assert middleware.commit_on_exit is False # Default value
- finally:
- asyncio.run(custom_engine.dispose())
diff --git a/tests/test_coverage_improvements.py b/tests/test_coverage_improvements.py
deleted file mode 100644
index b94f755..0000000
--- a/tests/test_coverage_improvements.py
+++ /dev/null
@@ -1,287 +0,0 @@
-"""
-Tests to improve code coverage for edge cases and fallback imports.
-"""
-
-import asyncio
-from unittest.mock import MagicMock, patch
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-from sqlalchemy.sql import text
-
-from fastapi_async_sqlalchemy import SQLAlchemyMiddleware, db
-
-
-@pytest.mark.asyncio
-async def test_cleanup_callback_with_closed_loop():
- """Test cleanup callback when event loop is closed."""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_closed_loop")
- async def test_closed_loop():
- async with db(multi_sessions=True):
- # Create a task that will trigger cleanup
- async def child_task():
- session = db.session
- await session.execute(text("SELECT 1"))
- return "done"
-
- task = asyncio.create_task(child_task())
- result = await task
-
- return {"result": result}
-
- with TestClient(app) as client:
- response = client.get("/test_closed_loop")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_cleanup_callback_runtime_error():
- """Test cleanup callback when get_running_loop raises RuntimeError."""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_runtime_error")
- async def test_runtime_error():
- async with db(multi_sessions=True):
-
- async def child_task():
- session = db.session
- await session.execute(text("SELECT 42"))
- return "ok"
-
- task = asyncio.create_task(child_task())
- result = await task
-
- # Give time for cleanup callbacks to execute
- await asyncio.sleep(0.1)
- return {"result": result}
-
- with TestClient(app) as client:
- response = client.get("/test_runtime_error")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_multiple_child_tasks_cleanup():
- """Test that multiple child tasks all get cleanup callbacks."""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_multiple_cleanup")
- async def test_multiple_cleanup():
- results = []
- async with db(multi_sessions=True):
-
- async def child_task(n):
- session = db.session
- result = await session.execute(text(f"SELECT {n}"))
- return result.scalar()
-
- tasks = [asyncio.create_task(child_task(i)) for i in range(5)]
- results = await asyncio.gather(*tasks)
-
- return {"results": results}
-
- with TestClient(app) as client:
- response = client.get("/test_multiple_cleanup")
- assert response.status_code == 200
- assert len(response.json()["results"]) == 5
-
-
-def test_import_coverage_markers():
- """Test that import fallback code paths are marked for coverage."""
- # This test ensures that import fallback blocks are properly marked
- # even if they can't be executed in the current environment
-
- # The actual imports happen at module load time, so we can't test them
- # directly without manipulating sys.modules before import.
- # Instead, we verify the code exists and is syntactically correct.
-
- import fastapi_async_sqlalchemy.middleware as middleware_module
-
- # Verify that DefaultAsyncSession is set
- assert hasattr(middleware_module, "DefaultAsyncSession")
-
- # Check if we're using SQLModel or plain AsyncSession
- from sqlalchemy.ext.asyncio import AsyncSession
-
- assert issubclass(middleware_module.DefaultAsyncSession, AsyncSession)
-
-
-@pytest.mark.asyncio
-async def test_current_task_none_scenario():
- """
- Test scenario where asyncio.current_task() might return None.
-
- Note: This is extremely rare in practice and hard to reproduce,
- as current_task() only returns None when called outside an event loop.
- The middleware already requires an event loop, so this edge case
- is primarily for completeness.
- """
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_task_context")
- async def test_task_context():
- # Verify we're in a task context
- task = asyncio.current_task()
- assert task is not None, "Should have current task in request context"
-
- async with db(multi_sessions=True):
- # Access session within task context
- session = db.session
- await session.execute(text("SELECT 1"))
-
- return {"success": True}
-
- with TestClient(app) as client:
- response = client.get("/test_task_context")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_edge_case_loop_closing_during_cleanup():
- """Test edge case where loop closes during cleanup callback setup."""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_loop_edge")
- async def test_loop_edge():
- async with db(multi_sessions=True, commit_on_exit=True):
- # Create multiple tasks that will all need cleanup
- async def quick_task(n):
- s = db.session
- await s.execute(text(f"SELECT {n}"))
-
- tasks = [asyncio.create_task(quick_task(i)) for i in range(3)]
- await asyncio.gather(*tasks)
-
- return {"done": True}
-
- with TestClient(app) as client:
- response = client.get("/test_loop_edge")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_current_task_none_with_mock():
- """
- Test behavior when current_task() returns None (sync context fallback).
-
- After the backward compatibility fix, this now creates a session for sync context
- instead of raising RuntimeError.
- """
-
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_none_task")
- async def test_none_task():
- # Temporarily mock current_task to return None
- with patch("asyncio.current_task", return_value=None):
- async with db(multi_sessions=True):
- # After backward compatibility fix, this should work
- # by falling back to sync context session
- session = db.session
- if session is not None:
- return {"success": True, "has_session": True}
- return {"error": "Session is None"}
-
- with TestClient(app) as client:
- response = client.get("/test_none_task")
- assert response.status_code == 200
- assert response.json()["success"] is True
- assert response.json()["has_session"] is True
-
-
-@pytest.mark.asyncio
-async def test_cleanup_callback_with_mocked_closed_loop():
- """
- Test cleanup callback behavior when loop.is_closed() returns True.
-
- This is a direct test of the cleanup_callback function to ensure
- it handles the closed loop case properly (lines 109-110).
- """
-
- # We need to test the cleanup callback directly
- # First, let's access the session property to trigger session creation with cleanup
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- callback_executed = []
-
- @app.get("/test_mock_closed")
- async def test_mock_closed():
- async with db(multi_sessions=True):
- # Patch get_running_loop inside the cleanup callback
-
- async def child_with_mock():
- session = db.session
- await session.execute(text("SELECT 1"))
-
- # After this task finishes, the cleanup callback will be called
- # We want to mock get_running_loop at that point
- return "done"
-
- # Patch asyncio.get_running_loop to return a closed loop
- def mock_get_running_loop_closed():
- loop = MagicMock()
- loop.is_closed.return_value = True
- callback_executed.append("closed_loop_path")
- return loop
-
- with patch("asyncio.get_running_loop", side_effect=mock_get_running_loop_closed):
- task = asyncio.create_task(child_with_mock())
- await task
- # Task is done, cleanup callback will execute with mocked get_running_loop
-
- return {"done": True}
-
- with TestClient(app) as client:
- with pytest.warns(UserWarning, match="No running event loop during cleanup"):
- response = client.get("/test_mock_closed")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_cleanup_callback_with_runtime_error():
- """
- Test cleanup callback when get_running_loop() raises RuntimeError.
-
- This tests the except RuntimeError branch (lines 111-112).
- """
-
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- callback_executed = []
-
- @app.get("/test_runtime_error")
- async def test_runtime_error():
- async with db(multi_sessions=True):
-
- async def child_with_runtime_error():
- session = db.session
- await session.execute(text("SELECT 1"))
- return "done"
-
- # Patch get_running_loop to raise RuntimeError
- def mock_get_running_loop_error():
- callback_executed.append("runtime_error_path")
- raise RuntimeError("No running event loop")
-
- with patch("asyncio.get_running_loop", side_effect=mock_get_running_loop_error):
- task = asyncio.create_task(child_with_runtime_error())
- await task
- # Cleanup callback will execute and hit the RuntimeError path
-
- return {"done": True}
-
- with TestClient(app) as client:
- with pytest.warns(UserWarning, match="No running event loop during cleanup"):
- response = client.get("/test_runtime_error")
- assert response.status_code == 200
diff --git a/tests/test_edge_cases_coverage.py b/tests/test_edge_cases_coverage.py
index 4a9155b..fa09cc7 100644
--- a/tests/test_edge_cases_coverage.py
+++ b/tests/test_edge_cases_coverage.py
@@ -4,16 +4,15 @@
"""
import asyncio
+from unittest.mock import patch
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy.exc import SQLAlchemyError
-from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.sql import text
from fastapi_async_sqlalchemy import SQLAlchemyMiddleware, db
-from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
@pytest.mark.asyncio
@@ -160,86 +159,6 @@ async def tracking_rollback():
assert response.status_code == 500 or response.status_code == 200
-@pytest.mark.asyncio
-async def test_session_created_without_tracking_warning():
- """Test warning when session is created without tracking (lines 117-122)"""
- # This is tricky to test as it requires accessing session property
- # outside of proper context setup
-
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- SQLAlchemyMiddleware_local, db_local = create_middleware_and_session_proxy()
-
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware_local, db_url="sqlite+aiosqlite:///:memory:")
-
- with TestClient(app):
- pass
-
- # This test verifies the warning path exists
- # In normal usage, the tracking set is always created in __aenter__
- # so this warning shouldn't occur in production
-
-
-def test_custom_engine_branch():
- """Test that custom_engine branch is exercised (line 61)"""
- SQLAlchemyMiddleware_local, db_local = create_middleware_and_session_proxy()
-
- app = FastAPI()
-
- # Create custom engine
- custom_engine = create_async_engine("sqlite+aiosqlite:///:memory:")
-
- try:
- # This should use the else branch on line 61
- middleware = SQLAlchemyMiddleware_local(
- app,
- custom_engine=custom_engine,
- commit_on_exit=False,
- )
-
- assert middleware is not None
- assert middleware.commit_on_exit is False
- finally:
- asyncio.run(custom_engine.dispose())
-
-
-@pytest.mark.asyncio
-async def test_import_fallback_coverage():
- """
- Test to document import fallback behavior (lines 18-19, 26-27)
- These lines are only executed in environments without SQLAlchemy 2.0+
- or without SQLModel installed
- """
- # Line 18-19: async_sessionmaker fallback
- # This is only needed for SQLAlchemy < 2.0
- # In modern SQLAlchemy (2.0+), async_sessionmaker exists
-
- try:
- from sqlalchemy.ext.asyncio import async_sessionmaker
-
- assert async_sessionmaker is not None
- # If we're here, lines 18-19 won't execute
- except ImportError: # pragma: no cover
- # In older SQLAlchemy, this would execute
- from sqlalchemy.orm import sessionmaker
-
- assert sessionmaker is not None
-
- # Lines 26-27: SQLModel fallback
- # These lines execute when SQLModel is NOT installed
- try:
- from sqlmodel.ext.asyncio.session import AsyncSession as SQLModelAsyncSession
-
- # If SQLModel is available, line 27 won't execute
- assert SQLModelAsyncSession is not None
- except ImportError:
- # Line 27 would execute if SQLModel not available
- from sqlalchemy.ext.asyncio import AsyncSession
-
- assert AsyncSession is not None
-
-
@pytest.mark.asyncio
async def test_multi_session_cleanup_all_paths():
"""Comprehensive test for all multi-session cleanup paths"""
@@ -309,3 +228,38 @@ async def test_single_exception():
with TestClient(app) as client:
response = client.get("/test_single_exception")
assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_cleanup_callback_without_running_loop_warns():
+ """Cleanup callback must warn (not crash) when no event loop is available.
+
+ Covers the `except RuntimeError` fallback when capturing the loop at
+ session creation time and the "No running event loop" warning in the
+ task-done cleanup callback.
+ """
+ app = FastAPI()
+ app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
+
+ @app.get("/test_no_loop")
+ async def test_no_loop():
+ async with db(multi_sessions=True):
+
+ async def child_task():
+ session = db.session
+ await session.execute(text("SELECT 1"))
+ return "done"
+
+ with patch(
+ "asyncio.get_running_loop",
+ side_effect=RuntimeError("No running event loop"),
+ ):
+ task = asyncio.create_task(child_task())
+ await task
+
+ return {"done": True}
+
+ with TestClient(app) as client:
+ with pytest.warns(UserWarning, match="No running event loop during cleanup"):
+ response = client.get("/test_no_loop")
+ assert response.status_code == 200
diff --git a/tests/test_import_fallback_simulation.py b/tests/test_import_fallback_simulation.py
deleted file mode 100644
index 848fe62..0000000
--- a/tests/test_import_fallback_simulation.py
+++ /dev/null
@@ -1,200 +0,0 @@
-"""
-Tests to verify import fallback behavior
-These tests document the behavior of lines 18-19 and 26-27
-which only execute in specific import scenarios
-"""
-
-import asyncio
-
-import pytest
-
-
-def test_async_sessionmaker_import_documentation():
- """
- Document async_sessionmaker import fallback (lines 18-19)
-
- Lines 18-19 in middleware.py:
- except ImportError:
- from sqlalchemy.orm import sessionmaker as async_sessionmaker
-
- These lines only execute when SQLAlchemy doesn't have async_sessionmaker,
- which would be SQLAlchemy < 2.0. Since our project requires SQLAlchemy 1.4.19+,
- this fallback ensures compatibility.
-
- In modern SQLAlchemy (2.0+), async_sessionmaker exists, so line 18-19 won't run.
- """
- # Verify that async_sessionmaker is available in current environment
- from sqlalchemy.ext.asyncio import async_sessionmaker
-
- assert async_sessionmaker is not None
- assert callable(async_sessionmaker)
-
-
-def test_sqlmodel_import_documentation():
- """
- Document SQLModel AsyncSession import fallback (lines 26-27)
-
- Lines 26-27 in middleware.py:
- except ImportError:
- DefaultAsyncSession: Type[AsyncSession] = AsyncSession
-
- Line 27 only executes when SQLModel is NOT installed.
- Since our test environment has SQLModel, line 27 won't be covered.
-
- This test documents that the fallback exists for environments without SQLModel.
- """
- # Check if SQLModel is available
- try:
- from sqlmodel.ext.asyncio.session import AsyncSession as SQLModelAsyncSession
-
- # SQLModel is available, so line 27 won't execute
- assert SQLModelAsyncSession is not None
-
- # Verify that our middleware uses SQLModel's AsyncSession
- from fastapi_async_sqlalchemy.middleware import DefaultAsyncSession
-
- assert DefaultAsyncSession == SQLModelAsyncSession
- except ImportError:
- # If SQLModel is not available, line 27 would execute
- from sqlalchemy.ext.asyncio import AsyncSession
-
- from fastapi_async_sqlalchemy.middleware import DefaultAsyncSession
-
- assert DefaultAsyncSession == AsyncSession
-
-
-def test_custom_engine_else_branch_execution():
- """
- Test to verify custom_engine else branch (line 61)
-
- The middleware has this structure:
- if not custom_engine:
- engine = create_async_engine(db_url, **engine_args)
- else:
- engine = custom_engine # Line 61
-
- We need to ensure this branch is actually executed.
- """
- from fastapi import FastAPI
- from sqlalchemy.ext.asyncio import create_async_engine
-
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- SQLAlchemyMiddleware_local, db_local = create_middleware_and_session_proxy()
-
- app = FastAPI()
-
- # Create a custom engine with specific settings
- custom_engine = create_async_engine(
- "sqlite+aiosqlite:///:memory:", echo=False, pool_pre_ping=True
- )
-
- try:
- # Initialize middleware with custom_engine
- # This should execute line 61: engine = custom_engine
- middleware = SQLAlchemyMiddleware_local(app, custom_engine=custom_engine)
-
- # Verify middleware was created
- assert middleware is not None
- finally:
- asyncio.run(custom_engine.dispose())
-
-
-def test_session_tracking_warning_scenario():
- """
- Test the warning scenario on line 117
-
- This warning occurs when:
- - multi_sessions mode is active
- - A session is created (via db.session property)
- - But _tracked_sessions.get() returns None
-
- This should not happen in normal usage since __aenter__ sets up tracking,
- but the warning is there as a safety check.
- """
-
- # This tests that the code path exists
- # In practice, the tracking set is always created in __aenter__
- # before any session can be accessed
-
- # The warning would appear if somehow the tracking context var was not set
- # when accessing db.session in multi_sessions mode
-
- # Since this requires internal manipulation of context vars,
- # we document it here rather than trying to force the condition
-
-
-@pytest.mark.asyncio
-async def test_simulated_import_fallback_for_older_sqlalchemy():
- """
- Simulation test showing what would happen with older SQLAlchemy
-
- This test documents the behavior but cannot force the import
- without breaking the current environment.
- """
- # In an environment with SQLAlchemy < 2.0:
- # - Line 17 would fail to import async_sessionmaker
- # - Lines 18-19 would execute instead
- # - The middleware would use sessionmaker from sqlalchemy.orm
-
- # Since we're testing with SQLAlchemy 2.0+, we just verify
- # that the modern import works
- from sqlalchemy.ext.asyncio import async_sessionmaker
-
- assert async_sessionmaker is not None
-
-
-@pytest.mark.asyncio
-async def test_verify_all_middleware_branches_tested():
- """
- Meta-test to verify we've covered all major code paths
- """
- from fastapi import FastAPI
- from fastapi.testclient import TestClient
- from sqlalchemy.ext.asyncio import create_async_engine
-
- from fastapi_async_sqlalchemy import SQLAlchemyMiddleware
-
- # Test 1: db_url path (line 59: engine = create_async_engine)
- app1 = FastAPI()
- app1.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite://")
- with TestClient(app1) as client1:
- assert client1 is not None
-
- # Test 2: custom_engine path (line 61: engine = custom_engine)
- from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
- SQLAlchemyMiddleware2, _ = create_middleware_and_session_proxy()
- app2 = FastAPI()
- custom_engine = create_async_engine("sqlite+aiosqlite://")
- app2.add_middleware(SQLAlchemyMiddleware2, custom_engine=custom_engine)
- try:
- with TestClient(app2) as client2:
- assert client2 is not None
- finally:
- await custom_engine.dispose()
-
-
-def test_coverage_report_explanation():
- """
- Documentation of remaining uncovered lines and why
-
- Uncovered Lines:
- - Lines 18-19: Import fallback for SQLAlchemy < 2.0
- Cannot be covered when running tests with SQLAlchemy 2.0+
-
- - Lines 26-27: Import fallback when SQLModel not installed
- Cannot be covered when running tests with SQLModel installed
-
- - Line 61: else branch for custom_engine
- Should be covered by custom_engine tests
-
- - Line 117: Warning for missing session tracking
- Defensive code that shouldn't occur in normal usage
-
- These lines provide important fallback and safety behavior
- but are difficult or impossible to cover in a test environment
- that has all dependencies installed.
- """
- # This test passes to document the coverage situation
- assert True
diff --git a/tests/test_import_fallbacks.py b/tests/test_import_fallbacks.py
deleted file mode 100644
index 7bb1838..0000000
--- a/tests/test_import_fallbacks.py
+++ /dev/null
@@ -1,58 +0,0 @@
-"""
-Tests for import fallback scenarios.
-
-These tests verify that the code handles missing optional dependencies gracefully.
-"""
-
-
-def test_sqlmodel_not_installed_fallback():
- """Test fallback when SQLModel is not installed."""
- import inspect
-
- import fastapi_async_sqlalchemy.middleware as mod
-
- # Verify the fallback code structure exists
- source = inspect.getsource(mod)
- assert "try:" in source
- assert "from sqlmodel.ext.asyncio.session import AsyncSession as SQLModelAsyncSession" in source
- assert "except ImportError:" in source
- assert "DefaultAsyncSession: type[AsyncSession] = AsyncSession" in source
-
-
-def test_default_async_session_type():
- """Test that DefaultAsyncSession is properly set."""
- from sqlalchemy.ext.asyncio import AsyncSession
-
- from fastapi_async_sqlalchemy.middleware import DefaultAsyncSession
-
- # Should be either SQLModel's AsyncSession or SQLAlchemy's AsyncSession
- assert issubclass(DefaultAsyncSession, AsyncSession)
-
- # Verify it's a valid session class
- assert hasattr(DefaultAsyncSession, "__init__")
-
-
-def test_coverage_pragmas_not_needed():
- """
- Verify that fallback imports don't need pragma: no cover.
-
- We achieve this by having tests that at least verify the code structure,
- even if we can't execute both paths in a single test run.
- """
- import inspect
-
- import fastapi_async_sqlalchemy.middleware as mod
-
- source = inspect.getsource(mod)
-
- # Ensure no pragma: no cover on import blocks
- # (These should be covered by structural tests)
- lines = source.split("\n")
- for i, line in enumerate(lines):
- if "pragma: no cover" in line:
- # Check if it's in an import block
- if i > 0 and "import" in lines[i - 1]:
- raise AssertionError(
- f"Import fallback at line {i} should not have pragma: no cover. "
- "Use structural tests instead."
- )
diff --git a/tests/test_maximum_coverage.py b/tests/test_maximum_coverage.py
deleted file mode 100644
index 8391d27..0000000
--- a/tests/test_maximum_coverage.py
+++ /dev/null
@@ -1,430 +0,0 @@
-"""
-Comprehensive tests to achieve maximum code coverage
-Focuses on uncovered lines in middleware.py
-"""
-
-import asyncio
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-from sqlalchemy.exc import SQLAlchemyError
-from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
-from sqlalchemy.sql import text
-
-from fastapi_async_sqlalchemy import SQLAlchemyMiddleware, db
-from fastapi_async_sqlalchemy.exceptions import MissingSessionError, SessionNotInitialisedError
-from fastapi_async_sqlalchemy.middleware import create_middleware_and_session_proxy
-
-
-@pytest.mark.asyncio
-async def test_multi_session_cleanup_with_commit_exception():
- """Commit failure in multi-session cleanup must fail the request."""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite://")
-
- @app.get("/test_commit_failure")
- async def test_commit_failure():
- async with db(multi_sessions=True, commit_on_exit=True):
- # Access session to trigger creation
- session = db.session
-
- # Mock the commit to raise an exception
- async def failing_commit():
- raise SQLAlchemyError("Simulated commit failure")
-
- session.commit = failing_commit
-
- # Store original rollback to verify it was called
- rollback_called = False
- original_rollback = session.rollback
-
- async def tracking_rollback():
- nonlocal rollback_called
- rollback_called = True
- await original_rollback()
-
- session.rollback = tracking_rollback
-
- return {"session_id": id(session)}
-
- with TestClient(app, raise_server_exceptions=False) as client:
- response = client.get("/test_commit_failure")
- assert response.status_code == 500
-
-
-@pytest.mark.asyncio
-async def test_multi_session_commit_on_exit_success():
- """Test successful commit in multi-session mode with commit_on_exit=True"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- commit_was_called = False
-
- @app.get("/test_commit_success")
- async def test_commit_success():
- nonlocal commit_was_called
- async with db(multi_sessions=True, commit_on_exit=True):
- session = db.session
-
- # Track if commit is called
- original_commit = session.commit
-
- async def tracking_commit():
- nonlocal commit_was_called
- commit_was_called = True
- await original_commit()
-
- session.commit = tracking_commit
-
- # Execute a simple query
- await session.execute(text("SELECT 1"))
-
- return {"status": "ok"}
-
- with TestClient(app) as client:
- response = client.get("/test_commit_success")
- assert response.status_code == 200
-
- # Give cleanup time to run
- await asyncio.sleep(0.1)
-
-
-@pytest.mark.asyncio
-async def test_multi_session_multiple_tasks_with_cleanup():
- """Test multi-session mode with multiple concurrent tasks and verify cleanup"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- session_ids = []
-
- @app.get("/test_multi_cleanup")
- async def test_multi_cleanup():
- async with db(multi_sessions=True, commit_on_exit=True):
-
- async def execute_with_session(value: int):
- session = db.session
- session_ids.append(id(session))
- result = await session.execute(text(f"SELECT {value}"))
- return result.scalar()
-
- # Create multiple tasks
- tasks = [asyncio.create_task(execute_with_session(i)) for i in range(5)]
-
- results = await asyncio.gather(*tasks)
- return {"results": results, "session_count": len(set(session_ids))}
-
- with TestClient(app) as client:
- response = client.get("/test_multi_cleanup")
- assert response.status_code == 200
-
-
-def test_import_fallback_async_sessionmaker():
- """Test import fallback for async_sessionmaker (lines 18-19)"""
- # This test verifies the import works
- # The fallback is only used on older SQLAlchemy versions
- try:
- from sqlalchemy.ext.asyncio import async_sessionmaker
-
- assert async_sessionmaker is not None
- except ImportError: # pragma: no cover
- # If async_sessionmaker doesn't exist, the fallback should work
- from sqlalchemy.orm import sessionmaker
-
- assert sessionmaker is not None
-
-
-def test_import_fallback_sqlmodel():
- """Test import fallback for SQLModel AsyncSession (lines 26-27)"""
- # Test that DefaultAsyncSession is properly set
- from fastapi_async_sqlalchemy.middleware import DefaultAsyncSession
-
- # It should be a subclass of AsyncSession regardless of SQLModel availability
- assert issubclass(DefaultAsyncSession, AsyncSession)
-
- # Check if SQLModel is available
- try:
- from sqlmodel.ext.asyncio.session import AsyncSession as SQLModelAsyncSession
-
- # If SQLModel is available, DefaultAsyncSession should be SQLModelAsyncSession
- assert DefaultAsyncSession == SQLModelAsyncSession
- except ImportError:
- # If SQLModel is not available, DefaultAsyncSession should be regular AsyncSession
- assert DefaultAsyncSession == AsyncSession
-
-
-def test_db_url_none_validation():
- """Test line 58: db_url validation when it's explicitly None"""
- # This is actually unreachable code due to line 54-55 check
- # But we can verify the validation logic
- SQLAlchemyMiddleware_local, _ = create_middleware_and_session_proxy()
-
- app = FastAPI()
-
- # This should raise ValueError at line 55
- with pytest.raises(ValueError, match="You need to pass a db_url or a custom_engine parameter"):
- SQLAlchemyMiddleware_local(app, db_url=None, custom_engine=None)
-
-
-def test_custom_engine_path():
- """Test middleware initialization with custom_engine (line 61)"""
- SQLAlchemyMiddleware_local, db_local = create_middleware_and_session_proxy()
-
- app = FastAPI()
- custom_engine = create_async_engine("sqlite+aiosqlite:///:memory:")
-
- try:
- # Initialize with custom engine
- middleware = SQLAlchemyMiddleware_local(app, custom_engine=custom_engine)
- assert middleware.commit_on_exit is False
- finally:
- asyncio.run(custom_engine.dispose())
-
- # Verify it doesn't require db_url
- # This covers the else branch on line 61
-
-
-@pytest.mark.asyncio
-async def test_session_outside_middleware_context():
- """Test accessing session outside middleware raises MissingSessionError"""
- # Create a fresh middleware instance
- SQLAlchemyMiddleware_local, db_local = create_middleware_and_session_proxy()
-
- app = FastAPI()
- SQLAlchemyMiddleware_local(app, db_url="sqlite+aiosqlite://")
-
- # Try to access session outside of request context
- with pytest.raises(MissingSessionError):
- _ = db_local.session
-
-
-@pytest.mark.asyncio
-async def test_multi_session_mode_context_vars():
- """Test that multi_session mode properly sets and resets context variables"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_context_vars")
- async def test_context_vars():
- async with db(multi_sessions=True, commit_on_exit=True):
- session1 = db.session
- session2 = db.session
-
- assert session1 is not None
- assert session2 is not None
- assert session1 is session2
-
- return {"status": "ok"}
-
- with TestClient(app) as client:
- response = client.get("/test_context_vars")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_regular_session_context_exit_with_exception():
- """Test that regular session mode rolls back on exception (line 162)"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_rollback")
- async def test_rollback():
- try:
- async with db():
- session = db.session
- await session.execute(text("SELECT 1"))
- # Simulate an error
- raise ValueError("Test exception")
- except ValueError:
- pass
-
- return {"status": "rolled_back"}
-
- with TestClient(app) as client:
- response = client.get("/test_rollback")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_regular_session_commit_on_exit():
- """Test regular session mode with commit_on_exit=True (line 164)"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_commit")
- async def test_commit():
- async with db(commit_on_exit=True):
- session = db.session
- await session.execute(text("SELECT 1"))
- # No exception, should commit
-
- return {"status": "committed"}
-
- with TestClient(app) as client:
- response = client.get("/test_commit")
- assert response.status_code == 200
-
-
-def test_middleware_commit_on_exit_parameter():
- """Test SQLAlchemyMiddleware with commit_on_exit parameter"""
- SQLAlchemyMiddleware_local, db_local = create_middleware_and_session_proxy()
-
- app = FastAPI()
-
- # Test with commit_on_exit=True
- middleware = SQLAlchemyMiddleware_local(app, db_url="sqlite+aiosqlite://", commit_on_exit=True)
- assert middleware.commit_on_exit is True
-
- # Test with commit_on_exit=False on a separate proxy to avoid rebinding the singleton.
- SecondMiddleware, _ = create_middleware_and_session_proxy()
- middleware2 = SecondMiddleware(app, db_url="sqlite+aiosqlite://", commit_on_exit=False)
- assert middleware2.commit_on_exit is False
-
-
-def test_engine_args_and_session_args():
- """Test middleware initialization with engine_args and session_args"""
- SQLAlchemyMiddleware_local, db_local = create_middleware_and_session_proxy()
-
- app = FastAPI()
-
- # Use valid engine args for sqlite
- engine_args = {"echo": True}
- # Don't pass expire_on_commit since it's already set to False in middleware
- session_args = {"autoflush": False}
-
- middleware = SQLAlchemyMiddleware_local(
- app, db_url="sqlite+aiosqlite://", engine_args=engine_args, session_args=session_args
- )
-
- assert middleware is not None
-
-
-@pytest.mark.asyncio
-async def test_session_not_initialised_in_context():
- """Test SessionNotInitialisedError in __aenter__ (line 145)"""
- # Create a fresh instance without initializing
- SQLAlchemyMiddleware_local, db_local = create_middleware_and_session_proxy()
-
- # Try to use context without initializing middleware
- with pytest.raises(SessionNotInitialisedError):
- async with db_local():
- pass
-
-
-@pytest.mark.asyncio
-async def test_multi_session_token_reset():
- """Test that multi_session tokens are properly reset (lines 156-157)"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_token_reset")
- async def test_token_reset():
- # Use multi_sessions context
- async with db(multi_sessions=True):
- session = db.session
- await session.execute(text("SELECT 1"))
-
- # After exiting, should not be in multi_sessions mode
- # Verify by trying to access session (should raise MissingSessionError)
- return {"status": "ok"}
-
- with TestClient(app) as client:
- response = client.get("/test_token_reset")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_session_args_parameter():
- """Test DBSession with session_args parameter"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_session_args")
- async def test_session_args():
- # Use session_args in context
- session_args = {"expire_on_commit": False}
- async with db(session_args=session_args):
- session = db.session
- result = await session.execute(text("SELECT 42"))
- value = result.scalar()
-
- return {"value": value}
-
- with TestClient(app) as client:
- response = client.get("/test_session_args")
- assert response.status_code == 200
- assert response.json()["value"] == 42
-
-
-@pytest.mark.asyncio
-async def test_multi_session_without_commit_on_exit():
- """Test multi_session mode with commit_on_exit=False (default)"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_no_commit")
- async def test_no_commit():
- async with db(multi_sessions=True, commit_on_exit=False):
- session = db.session
- await session.execute(text("SELECT 1"))
- # Should not commit on cleanup
-
- return {"status": "no_commit"}
-
- with TestClient(app) as client:
- response = client.get("/test_no_commit")
- assert response.status_code == 200
-
-
-@pytest.mark.asyncio
-async def test_task_done_callback_cleanup():
- """Test that cleanup is added as task done callback (line 122)"""
- app = FastAPI()
- app.add_middleware(SQLAlchemyMiddleware, db_url="sqlite+aiosqlite:///:memory:")
-
- @app.get("/test_callback")
- async def test_callback():
- async with db(multi_sessions=True, commit_on_exit=True):
-
- async def task_function():
- session = db.session
- await session.execute(text("SELECT 1"))
- return "done"
-
- # Create a task that will have cleanup callback
- task = asyncio.create_task(task_function())
- result = await task
-
- return {"result": result}
-
- with TestClient(app) as client:
- response = client.get("/test_callback")
- assert response.status_code == 200
-
- # Give cleanup time to execute
- await asyncio.sleep(0.1)
-
-
-def test_all_exception_classes():
- """Test all custom exception classes"""
- from fastapi_async_sqlalchemy.exceptions import (
- MissingSessionError,
- SessionNotInitialisedError,
- )
-
- # Test SessionNotInitialisedError
- exc1 = SessionNotInitialisedError()
- assert "not initialised" in str(exc1).lower()
- assert isinstance(exc1, Exception)
-
- # Test MissingSessionError
- exc2 = MissingSessionError()
- assert "no session found" in str(exc2).lower()
- assert isinstance(exc2, Exception)
-
- # Test that they can be raised
- with pytest.raises(SessionNotInitialisedError):
- raise SessionNotInitialisedError()
-
- with pytest.raises(MissingSessionError):
- raise MissingSessionError()
diff --git a/tests/test_type_hints_compatibility.py b/tests/test_type_hints_compatibility.py
index baad1cc..758e392 100644
--- a/tests/test_type_hints_compatibility.py
+++ b/tests/test_type_hints_compatibility.py
@@ -7,7 +7,6 @@
from fastapi_async_sqlalchemy import (
DBSessionMeta,
- DBSessionType,
db,
)
@@ -18,17 +17,6 @@ def test_dbsessionmeta_is_exported():
assert isinstance(DBSessionMeta, type)
-def test_dbsessiontype_is_exported():
- """Test that DBSessionType is available for import (alternative name)"""
- assert DBSessionType is not None
- assert isinstance(DBSessionType, type)
-
-
-def test_dbsessionmeta_equals_dbsessiontype():
- """Test that both names refer to the same type"""
- assert DBSessionMeta is DBSessionType
-
-
def test_type_of_db_is_dbsessionmeta():
"""Test that db instance has DBSessionMeta as its type"""
assert type(db) is DBSessionMeta
@@ -48,17 +36,6 @@ def get_db() -> DBSessionMeta:
assert type(result) is DBSessionMeta
-def test_alternative_type_hint_name():
- """Test that DBSessionType works as type hint"""
-
- def get_db_session() -> DBSessionType:
- return db
-
- result = get_db_session()
- assert result is db
- assert type(result) is DBSessionType
-
-
def test_backwards_compatibility_with_old_code():
"""
Test backwards compatibility with code from v0.5