FastAPI + ProcessPoolExecutor

main.py

import asyncio
from concurrent.futures import ProcessPoolExecutor
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
from time import time
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: "FastAPI"):
    max_workers = 5
    app.state.executor = ProcessPoolExecutor(max_workers=max_workers)
    print(f"Started ProcessPoolExecutor with max_workers={max_workers}")
    yield
    app.state.executor.shutdown()


async def run_in_process(executor, fn, *args):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, fn, *args)  # wait and return result

fastapi_app = FastAPI(docs_url="/api/docs", lifespan=lifespan)

...

@router.post("/avatar-upload/")
async def update_avatar(
    request: Request,
    img_base64: str = Body(..., description="Base64 image object"),
):
    # some logic with base64 image

    executor = request.app.state.executor
    res = await run_in_process(executor, time)
    return {'result': res}