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

fastapi-async-vs-non-blocking.py

from time import sleep
from fastapi import FastAPI
import logging


logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)

logger = logging.getLogger(__name__)

app = FastAPI()


async def do_async_work_to_prepare_for_the_upload():
    pass


async def upload_to_s3():
    await do_async_work_to_prepare_for_the_upload()
    # assume the upload will run synchronously for 5 seconds, such as when
    # using the official aws boto3 client that doesn't support async/await
    logger.info("Before sleep")
    sleep(5)
    logger.info("After sleep")


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