Limit number of background tasks

[sakshamgoyal-01]

Possibly related: #943

I am currently using a threadpoolexecutor to add my process queue and limit to max 3 workers at a time. Is there a convenient way to limit background tasks in starlette/fastapi?

The goal is to allow processing of file uploads in the background but limit to 3 at a time in a queue to prevent rate/capacity limits.

[ltianyi992]

Starlette's BackgroundTasks does not have built-in concurrency limiting — tasks are dispatched immediately after the response is sent and there is no queue or semaphore mechanism.

For limiting to N concurrent background tasks, the idiomatic approach is an asyncio.Semaphore:

import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, BackgroundTasks, UploadFile

_UPLOAD_SEM = asyncio.Semaphore(3)  # max 3 concurrent uploads

async def process_upload(file_bytes: bytes) -> None:
    async with _UPLOAD_SEM:  # blocks if 3 are already running
        await asyncio.sleep(2)  # simulate long processing
        print(f"Processed {len(file_bytes)} bytes")

@app.post("/upload")
async def upload(file: UploadFile, background_tasks: BackgroundTasks):
    data = await file.read()
    background_tasks.add_task(process_upload, data)
    return {"status": "queued"}

The semaphore limits concurrency: if 3 uploads are already running, the 4th coroutine blocks inside async with _UPLOAD_SEM until one finishes.

For a true bounded queue (reject when full rather than block):

_UPLOAD_SEM = asyncio.Semaphore(3)

async def process_upload(file_bytes: bytes) -> None:
    if _UPLOAD_SEM.locked() and _UPLOAD_SEM._value == 0:
        return  # drop task; or raise / return 429 in the endpoint instead
    async with _UPLOAD_SEM:
        ...  # do work

Or check capacity before adding to BackgroundTasks and return a 429 Too Many Requests response.

For your thread-pool approach, wrapping ThreadPoolExecutor(max_workers=3) with loop.run_in_executor and submitting synchronous work is also fine for CPU-bound or blocking-IO processing:

from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=3)

def blocking_process(data: bytes) -> None:
    ...

@app.post("/upload")
async def upload(file: UploadFile, background_tasks: BackgroundTasks):
    data = await file.read()
    loop = asyncio.get_event_loop()
    background_tasks.add_task(loop.run_in_executor, executor, blocking_process, data)
    return {"status": "queued"}

The ThreadPoolExecutor with max_workers=3 naturally limits concurrency for sync work.

[Attakay78]

@sakshamgoyal-01

For managing background tasks in fastapi you can checkout this third party library https://github.com/Attakay78/fastapi-taskflow. It has built in concurrency support.