Code snippets showing different approaches to resolving blocking operations in a FastAPI server.

fastapi-async-vs-non-blocking.py

...

async def upload_to_s3():
    await do_async_work_to_prepare_for_the_upload()
    # assume the upload will run asynchronously for 5 seconds, such as
    # when using aioboto3 (unofficial async-compatible aws sdk)
    logger.info("Before sleep")
    await asyncio.sleep(5)
    logger.info("After sleep")


@app.get("/process-files")
async def process_files(background_tasks: BackgroundTasks):
    logger.info("Before process files")
    background_tasks.add_task(upload_to_s3)
    logger.info("After process files")
    return {"ok": True}