Browse Source

Fix issue #11215: Merge background tasks from dependency injection with Response.background

When a user returns a Response with a background parameter and also uses
injected BackgroundTasks, the injected tasks were being silently discarded.
This fix merges both sets of background tasks so all tasks are executed.

Changes:
- Modified fastapi/routing.py to merge BackgroundTasks when both exist
- Added comprehensive tests in test_background_tasks_merge.py
- Handles both BackgroundTask and BackgroundTasks types
- Maintains backward compatibility

Fixes #11215
pull/14868/head
Varun Chawla 5 months ago
parent
commit
803cff2f2f
  1. 25
      fastapi/routing.py
  2. 111
      tests/test_background_tasks_merge.py

25
fastapi/routing.py

@ -431,6 +431,31 @@ def get_request_handler(
if isinstance(raw_response, Response): if isinstance(raw_response, Response):
if raw_response.background is None: if raw_response.background is None:
raw_response.background = solved_result.background_tasks raw_response.background = solved_result.background_tasks
elif solved_result.background_tasks is not None:
# Merge background tasks from dependency injection with tasks from Response
# The Response's background can be either a BackgroundTask or BackgroundTasks
from starlette.background import BackgroundTask, BackgroundTasks as StarletteBackgroundTasks
if isinstance(raw_response.background, StarletteBackgroundTasks):
# If Response has BackgroundTasks, prepend the injected tasks
if isinstance(solved_result.background_tasks, StarletteBackgroundTasks):
# Prepend all injected tasks before the response tasks
raw_response.background.tasks = (
solved_result.background_tasks.tasks + raw_response.background.tasks
)
elif isinstance(raw_response.background, BackgroundTask):
# If Response has a single BackgroundTask, convert to BackgroundTasks
response_task = raw_response.background
if isinstance(solved_result.background_tasks, StarletteBackgroundTasks):
# Create new BackgroundTasks with injected tasks + response task
raw_response.background = StarletteBackgroundTasks(
tasks=solved_result.background_tasks.tasks + [response_task]
)
else:
# Both are single tasks
raw_response.background = StarletteBackgroundTasks(
tasks=[solved_result.background_tasks, response_task]
)
response = raw_response response = raw_response
else: else:
response_args: dict[str, Any] = { response_args: dict[str, Any] = {

111
tests/test_background_tasks_merge.py

@ -0,0 +1,111 @@
"""
Test that background tasks from injected BackgroundTasks and Response.background are both executed.
Related to issue #11215
"""
from fastapi import BackgroundTasks, FastAPI
from fastapi.responses import Response
from fastapi.testclient import TestClient
app = FastAPI()
# Track which background tasks were executed
executed_tasks = []
def task_from_injected():
executed_tasks.append("injected")
def task_from_response():
executed_tasks.append("response")
@app.get("/both-tasks")
async def endpoint_with_both_tasks(tasks: BackgroundTasks):
"""Endpoint that uses both injected BackgroundTasks and Response.background"""
tasks.add_task(task_from_injected)
return Response(
content="Custom response",
background=BackgroundTasks(tasks=[]),
)
@app.get("/both-tasks-populated")
async def endpoint_with_both_tasks_populated(tasks: BackgroundTasks):
"""Endpoint where both injected and response have background tasks"""
tasks.add_task(task_from_injected)
response_bg = BackgroundTasks()
response_bg.add_task(task_from_response)
return Response(
content="Custom response",
background=response_bg,
)
@app.get("/only-injected")
async def endpoint_with_only_injected(tasks: BackgroundTasks):
"""Endpoint that only uses injected BackgroundTasks"""
tasks.add_task(task_from_injected)
return Response(content="Custom response")
@app.get("/only-response")
async def endpoint_with_only_response():
"""Endpoint that only uses Response.background"""
response_bg = BackgroundTasks()
response_bg.add_task(task_from_response)
return Response(
content="Custom response",
background=response_bg,
)
def test_injected_tasks_not_lost_when_response_has_empty_background():
"""Test that injected background tasks are preserved even when Response has empty background"""
global executed_tasks
executed_tasks = []
client = TestClient(app)
response = client.get("/both-tasks")
assert response.status_code == 200
# Both tasks should be executed
assert "injected" in executed_tasks, "Injected task was not executed"
def test_both_tasks_executed_when_both_populated():
"""Test that both injected and response background tasks are executed"""
global executed_tasks
executed_tasks = []
client = TestClient(app)
response = client.get("/both-tasks-populated")
assert response.status_code == 200
# Both tasks should be executed
assert "injected" in executed_tasks, "Injected task was not executed"
assert "response" in executed_tasks, "Response task was not executed"
def test_only_injected_tasks_work():
"""Test that only injected background tasks work when no Response.background"""
global executed_tasks
executed_tasks = []
client = TestClient(app)
response = client.get("/only-injected")
assert response.status_code == 200
assert "injected" in executed_tasks
def test_only_response_tasks_work():
"""Test that only response background tasks work"""
global executed_tasks
executed_tasks = []
client = TestClient(app)
response = client.get("/only-response")
assert response.status_code == 200
assert "response" in executed_tasks
Loading…
Cancel
Save