mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-09-19 19:29:55 +00:00
Compare commits
3
Commits
23dc5c86b2
...
6dafd40277
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6dafd40277 | ||
|
|
1c806cccd9
|
||
|
|
1354f59bb4 |
+1
-1
@@ -54,7 +54,7 @@ db = Database(
|
||||
|
||||
## Committing before the response
|
||||
|
||||
[`db.install(app)`](../reference/db.md#fastapi_toolsets.db.Database) adds a middleware that commits the request's session when the response starts, after the endpoint returns and before the body is sent. With the middleware installed, the dependency does not commit again.
|
||||
[`db.install(app)`](../reference/db.md#fastapi_toolsets.db.Database) adds a middleware that commits the request's session when the response starts, after the endpoint returns and before the body is sent. The dependency commits only if the middleware did not: when a function-scoped dependency unwinds before the response, or when the response never passes through the middleware. Either way the request is committed exactly once.
|
||||
|
||||
The request is committed as a single transaction:
|
||||
|
||||
|
||||
+1
-1
@@ -101,7 +101,7 @@ exclude = ["*.md"]
|
||||
extend-select = ["E712"]
|
||||
|
||||
[tool.ruff.lint.flake8-bugbear]
|
||||
extend-immutable-calls = ["fastapi.Depends"]
|
||||
extend-immutable-calls = ["fastapi.Depends", "fastapi.Security"]
|
||||
|
||||
[tool.ruff.lint.per-file-ignores]
|
||||
"tests/**" = ["RUF012", "RUF059", "SIM117", "DTZ001", "S110", "BLE001"]
|
||||
|
||||
@@ -160,9 +160,11 @@ def build_search_filters(
|
||||
column = field
|
||||
|
||||
# Build the filter (cast to String only when needed, to preserve
|
||||
# pg_trgm GIN index usability on already-String columns)
|
||||
# pg_trgm GIN index usability on already-String columns).
|
||||
column_as_string = (
|
||||
column if isinstance(column.type, String) else column.cast(String)
|
||||
column
|
||||
if isinstance(column.type, String) and not isinstance(column.type, Enum)
|
||||
else column.cast(String)
|
||||
)
|
||||
if config.case_sensitive:
|
||||
filters.append(column_as_string.like(f"%{query}%"))
|
||||
|
||||
@@ -66,10 +66,8 @@ class _CommitOnResponseMiddleware:
|
||||
|
||||
async def send_wrapper(message: Message) -> None:
|
||||
if message["type"] == "http.response.start":
|
||||
# ``scope["state"]`` is the same dict ``request.state`` writes
|
||||
# to, so this is the session stashed by the dependency.
|
||||
state = scope.get("state")
|
||||
session = state.get(self.state_attr) if state else None
|
||||
session = state.pop(self.state_attr, None) if state else None
|
||||
if session is not None and session.in_transaction():
|
||||
await session.commit()
|
||||
await send(message)
|
||||
@@ -158,7 +156,6 @@ class Database:
|
||||
# Private, per-instance state attribute; cannot collide with another
|
||||
# Database or be mismatched against the middleware.
|
||||
self._state_attr = f"_ft_db_session_{id(self):x}"
|
||||
self._middleware_installed = False
|
||||
self._disposed = False
|
||||
|
||||
async def _dispose(self) -> None:
|
||||
@@ -206,7 +203,6 @@ class Database:
|
||||
```
|
||||
"""
|
||||
app.add_middleware(_CommitOnResponseMiddleware, state_attr=self._state_attr)
|
||||
self._middleware_installed = True
|
||||
|
||||
inner_lifespan = app.router.lifespan_context
|
||||
|
||||
@@ -243,10 +239,17 @@ class Database:
|
||||
return await UserCrud.get(session, [User.id == user_id])
|
||||
```
|
||||
"""
|
||||
borrowed = getattr(request.state, self._state_attr, None)
|
||||
if borrowed is not None:
|
||||
yield borrowed
|
||||
return
|
||||
async with self._open() as session:
|
||||
setattr(request.state, self._state_attr, session)
|
||||
yield session
|
||||
if not self._middleware_installed and session.in_transaction():
|
||||
if (
|
||||
getattr(request.state, self._state_attr, None) is session
|
||||
and session.in_transaction()
|
||||
):
|
||||
await session.commit()
|
||||
|
||||
@asynccontextmanager
|
||||
|
||||
@@ -388,6 +388,76 @@ class TestBuildSearchFilters:
|
||||
|
||||
assert "CAST" in str(filters[0])
|
||||
|
||||
def test_casts_enum_column(self):
|
||||
"""Enum subclasses String but maps to a native DB enum, which has no ILIKE."""
|
||||
from fastapi_toolsets.crud.search import build_search_filters
|
||||
|
||||
filters, _ = build_search_filters(Order, "PEND", search_fields=[Order.status])
|
||||
|
||||
assert "CAST" in str(filters[0])
|
||||
|
||||
|
||||
class TestSearchEnumColumn:
|
||||
"""Searching an enum column must reach the database, not just build SQL."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_search_int_backed_enum(self, db_session: AsyncSession):
|
||||
"""Enum(int, Enum) stores names, so the cast makes 'PEND' match PENDING."""
|
||||
await OrderCrud.create(
|
||||
db_session, OrderCreate(name="a", status=OrderStatus.PENDING)
|
||||
)
|
||||
await OrderCrud.create(
|
||||
db_session, OrderCreate(name="b", status=OrderStatus.SHIPPED)
|
||||
)
|
||||
|
||||
result = await OrderCrud.offset_paginate(
|
||||
db_session,
|
||||
search="PEND",
|
||||
search_fields=[Order.status],
|
||||
schema=OrderRead,
|
||||
)
|
||||
|
||||
assert result.pagination.total_count == 1
|
||||
assert result.data[0].status is OrderStatus.PENDING
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_search_str_backed_enum(self, db_session: AsyncSession):
|
||||
"""Same for Enum(str, Enum) — still a native DB enum, still needs the cast."""
|
||||
await OrderCrud.create(
|
||||
db_session,
|
||||
OrderCreate(name="a", status=OrderStatus.PENDING, color=Color.BLUE),
|
||||
)
|
||||
await OrderCrud.create(
|
||||
db_session,
|
||||
OrderCreate(name="b", status=OrderStatus.PENDING, color=Color.RED),
|
||||
)
|
||||
|
||||
result = await OrderCrud.offset_paginate(
|
||||
db_session,
|
||||
search="BLU",
|
||||
search_fields=[Order.color],
|
||||
schema=OrderRead,
|
||||
)
|
||||
|
||||
assert result.pagination.total_count == 1
|
||||
assert result.data[0].color is Color.BLUE
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_search_mixed_enum_and_string_columns(self, db_session: AsyncSession):
|
||||
"""An enum column alongside a plain String column (the get_searchable_fields shape)."""
|
||||
await OrderCrud.create(
|
||||
db_session, OrderCreate(name="widget", status=OrderStatus.SHIPPED)
|
||||
)
|
||||
|
||||
result = await OrderCrud.offset_paginate(
|
||||
db_session,
|
||||
search="widget",
|
||||
search_fields=[Order.name, Order.status, Order.color],
|
||||
schema=OrderRead,
|
||||
)
|
||||
|
||||
assert result.pagination.total_count == 1
|
||||
|
||||
|
||||
class TestSearchConfig:
|
||||
"""Tests for SearchConfig options."""
|
||||
|
||||
+128
-9
@@ -6,7 +6,7 @@ from contextlib import asynccontextmanager
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import Depends, FastAPI
|
||||
from fastapi import Depends, FastAPI, Security
|
||||
from fastapi.responses import StreamingResponse
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from pydantic import PostgresDsn
|
||||
@@ -287,23 +287,37 @@ class TestDatabaseDependency:
|
||||
break
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_skips_commit_when_middleware_installed(self, engine, session_maker):
|
||||
"""With ``install()``, the dependency must NOT commit — the middleware owns it.
|
||||
async def test_second_resolution_borrows_session(self, engine):
|
||||
"""A second ``Depends(db)`` in one request reuses the stashed session."""
|
||||
db = Database(engine=engine)
|
||||
request = _make_request()
|
||||
|
||||
Here no middleware actually runs (we call the dependency directly), so the
|
||||
open transaction is rolled back on session close and nothing persists.
|
||||
"""
|
||||
owner_gen = db(request)
|
||||
owner = await anext(owner_gen)
|
||||
borrower_gen = db(request)
|
||||
assert await anext(borrower_gen) is owner
|
||||
|
||||
with pytest.raises(StopAsyncIteration): # teardown runs borrower-first
|
||||
await anext(borrower_gen)
|
||||
assert owner.in_transaction() # the borrower must not close what it borrowed
|
||||
|
||||
with pytest.raises(StopAsyncIteration):
|
||||
await anext(owner_gen)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_commits_when_middleware_did_not_run(self, engine, session_maker):
|
||||
"""``install()`` is per-``Database``, but the commit is per-request."""
|
||||
db = Database(engine=engine)
|
||||
db.install(FastAPI())
|
||||
|
||||
async for session in db(_make_request()):
|
||||
role = Role(name="mw_owns_commit")
|
||||
role = Role(name="mw_never_ran")
|
||||
session.add(role)
|
||||
await session.flush()
|
||||
|
||||
async with session_maker() as verify:
|
||||
result = await RoleCrud.first(verify, [Role.name == "mw_owns_commit"])
|
||||
assert result is None
|
||||
result = await RoleCrud.first(verify, [Role.name == "mw_never_ran"])
|
||||
assert result is not None
|
||||
|
||||
|
||||
class TestDatabaseSession:
|
||||
@@ -1523,6 +1537,55 @@ def _build_app(db: Database) -> FastAPI:
|
||||
await session.commit()
|
||||
return {"id": str(role.id), "name": role.name}
|
||||
|
||||
async def _scoped_writer(
|
||||
body: RoleCreate, session: AsyncSession = Security(db, scopes=["roles:write"])
|
||||
) -> int:
|
||||
# Security scopes give this a different dependency cache key than the
|
||||
# endpoint's plain ``Depends(db)``. Without borrowing it opens a second
|
||||
# session, and whichever one the middleware does not hold is discarded.
|
||||
await RoleCrud.create(session, RoleCreate(name=f"{body.name}_sub"))
|
||||
return id(session)
|
||||
|
||||
@app.post("/roles-two-cache-keys")
|
||||
async def create_via_two_cache_keys(
|
||||
body: RoleCreate,
|
||||
sub_session_id: int = Depends(_scoped_writer),
|
||||
session: AsyncSession = Depends(db),
|
||||
) -> dict:
|
||||
await RoleCrud.create(session, body)
|
||||
return {"same_session": sub_session_id == id(session)}
|
||||
|
||||
async def _fn_writer(
|
||||
body: RoleCreate, session: AsyncSession = Depends(db, scope="function")
|
||||
) -> None:
|
||||
# ``scope="function"`` unwinds before the response is sent, taking the
|
||||
# session with it — so the commit cannot be left to the middleware.
|
||||
await RoleCrud.create(session, RoleCreate(name=f"{body.name}_fn"))
|
||||
|
||||
@app.post("/roles-function-scope")
|
||||
async def create_with_function_scope(
|
||||
body: RoleCreate,
|
||||
boom: bool = False,
|
||||
_: None = Depends(_fn_writer),
|
||||
session: AsyncSession = Depends(db),
|
||||
) -> dict:
|
||||
await RoleCrud.create(session, body)
|
||||
if boom:
|
||||
raise RuntimeError("boom after write")
|
||||
return {"ok": True}
|
||||
|
||||
@app.post("/roles-function-scope-borrower")
|
||||
async def function_scope_borrows(
|
||||
body: RoleCreate,
|
||||
session: AsyncSession = Depends(db),
|
||||
_: None = Depends(_fn_writer),
|
||||
) -> dict:
|
||||
# Flipped order: the request-scoped dependency owns the session and the
|
||||
# function-scoped one borrows it. The borrower unwinds early but must not
|
||||
# commit or close — the commit still belongs to the middleware.
|
||||
await RoleCrud.create(session, body)
|
||||
return {"ok": True}
|
||||
|
||||
@app.get("/roles-stream/{name}")
|
||||
async def stream_role(
|
||||
name: str, session: AsyncSession = Depends(db)
|
||||
@@ -1619,6 +1682,62 @@ class TestCommitIntegration:
|
||||
# The write made before the stream began is durably committed.
|
||||
assert await _row_exists(session_maker, "streamed_role")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_two_cache_keys_share_one_session(self, engine, session_maker):
|
||||
"""Two resolutions of ``Depends(db)`` in one request must share a session."""
|
||||
app = _build_app(Database(engine=engine))
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
resp = await client.post("/roles-two-cache-keys", json={"name": "two_keys"})
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["same_session"] is True
|
||||
assert await _row_exists(session_maker, "two_keys")
|
||||
assert await _row_exists(session_maker, "two_keys_sub")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_function_scope_commits_before_response(self, engine, session_maker):
|
||||
"""``scope="function"`` unwinds before response-start, so the dependency
|
||||
commits on its way out instead of leaving it to the middleware."""
|
||||
app = _build_app(Database(engine=engine))
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
resp = await client.post("/roles-function-scope", json={"name": "fn_scope"})
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert await _row_exists(session_maker, "fn_scope")
|
||||
assert await _row_exists(session_maker, "fn_scope_fn")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_function_scope_borrower_leaves_commit_to_middleware(
|
||||
self, engine, session_maker
|
||||
):
|
||||
"""A function-scoped *borrower* unwinds early but owns nothing."""
|
||||
app = _build_app(Database(engine=engine))
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
resp = await client.post(
|
||||
"/roles-function-scope-borrower", json={"name": "fn_borrow"}
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert await _row_exists(session_maker, "fn_borrow")
|
||||
assert await _row_exists(session_maker, "fn_borrow_fn")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_function_scope_error_rolls_back(self, engine, session_maker):
|
||||
"""The early commit must still not fire when the request fails."""
|
||||
app = _build_app(Database(engine=engine))
|
||||
transport = ASGITransport(app=app, raise_app_exceptions=False)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
resp = await client.post(
|
||||
"/roles-function-scope?boom=true", json={"name": "fn_ghost"}
|
||||
)
|
||||
|
||||
assert resp.status_code == 500
|
||||
assert not await _row_exists(session_maker, "fn_ghost")
|
||||
assert not await _row_exists(session_maker, "fn_ghost_fn")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_multi_write_atomicity(self, engine, session_maker):
|
||||
"""When the 2nd write fails, the 1st must roll back too (one txn)."""
|
||||
|
||||
Reference in New Issue
Block a user