Emit UserWarning when Response.background silently discards injected BackgroundTasks

When an endpoint injects BackgroundTasks via dependency injection AND
returns a Response that already has its own `background` attribute set,
the injected tasks are silently dropped. This is a confusing footgun
that causes tasks to disappear without any indication.

This change adds a UserWarning in that scenario so the silent data loss
becomes visible. The warning message explains how to resolve the
conflict (either use the injected BackgroundTasks exclusively, or stop
injecting BackgroundTasks when using Response.background directly).

Ref: https://github.com/fastapi/fastapi/issues/11215

Co-Authored-By: Claude (claude-opus-4-6) <noreply@anthropic.com>
This commit is contained in:
easonysliu 2026-03-14 14:08:25 +08:00
parent 04b279fe77
commit 2ea91122c9
2 changed files with 112 additions and 0 deletions

View File

@ -4,6 +4,7 @@ import functools
import inspect
import json
import types
import warnings
from collections.abc import (
AsyncIterator,
Awaitable,
@ -675,6 +676,21 @@ def get_request_handler(
if isinstance(raw_response, Response):
if raw_response.background is None:
raw_response.background = solved_result.background_tasks
elif solved_result.background_tasks is not None:
warnings.warn(
"FastAPI has injected BackgroundTasks via "
"dependency injection, but the endpoint returned "
"a Response that already has a `background` set. "
"The dependency-injected background tasks will be "
"discarded in favor of the response's own "
"`background`. To avoid this warning, either add "
"your tasks to the injected `BackgroundTasks` "
"instance instead of setting `background` on the "
"response, or do not inject `BackgroundTasks` at "
"all.",
UserWarning,
stacklevel=1,
)
response = raw_response
else:
response_args = _build_response_args(

View File

@ -0,0 +1,96 @@
"""Test that a warning is emitted when a returned Response already has a background
task set and the dependency-injected BackgroundTasks would be silently discarded.
Ref: https://github.com/fastapi/fastapi/issues/11215
"""
import warnings
from fastapi import BackgroundTasks, FastAPI
from fastapi.testclient import TestClient
from starlette.background import BackgroundTask
from starlette.responses import JSONResponse, Response
def test_warn_when_response_background_overwrites_injected_tasks():
"""When the endpoint returns a Response with its own `background`,
and the user also injected `BackgroundTasks`, a UserWarning should
be emitted so the silent data loss is visible."""
app = FastAPI()
@app.get("/")
async def endpoint(tasks: BackgroundTasks):
tasks.add_task(lambda: None)
return Response(
content="Custom response",
background=BackgroundTask(lambda: None),
)
client = TestClient(app, raise_server_exceptions=False)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
resp = client.get("/")
assert resp.status_code == 200
bg_warnings = [
w for w in caught if "background" in str(w.message).lower()
]
assert len(bg_warnings) == 1
assert issubclass(bg_warnings[0].category, UserWarning)
def test_no_warn_when_response_has_no_background():
"""When the endpoint returns a Response without a `background`,
no warning should be emitted (the injected tasks are attached)."""
app = FastAPI()
executed = []
def bg_task():
executed.append(True)
@app.get("/")
async def endpoint(tasks: BackgroundTasks):
tasks.add_task(bg_task)
return JSONResponse(content={"ok": True})
client = TestClient(app)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
resp = client.get("/")
assert resp.status_code == 200
bg_warnings = [
w for w in caught if "background" in str(w.message).lower()
]
assert len(bg_warnings) == 0
# The injected task should have run
assert executed == [True]
def test_no_warn_when_no_injected_background_tasks():
"""When the endpoint returns a Response with a `background` but did
NOT inject BackgroundTasks, no warning should be emitted."""
app = FastAPI()
executed = []
@app.get("/")
async def endpoint():
return Response(
content="ok",
background=BackgroundTask(lambda: executed.append(True)),
)
client = TestClient(app)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
resp = client.get("/")
assert resp.status_code == 200
bg_warnings = [
w for w in caught if "background" in str(w.message).lower()
]
assert len(bg_warnings) == 0
assert executed == [True]