contextmanager_in_threadpool() — fastapi Function Reference

Architecture documentation for the contextmanager_in_threadpool() function in concurrency.py from the fastapi codebase.

Function python FastAPI Routing called by 1

Entity Profile

FastAPI→ Routing→ contextmanager_in_threadpool() — fastapi Function Reference

Dependency Diagram

graph TD
  b630faa9_9abe_ac01_eb1a_c46607435850["contextmanager_in_threadpool()"]
  e587e889_5c93_286d_8323_600fa8470a5c["concurrency.py"]
  b630faa9_9abe_ac01_eb1a_c46607435850 -->|defined in| e587e889_5c93_286d_8323_600fa8470a5c
  234ce98e_561d_d168_f588_50a82002939a["_solve_generator()"]
  234ce98e_561d_d168_f588_50a82002939a -->|calls| b630faa9_9abe_ac01_eb1a_c46607435850
  style b630faa9_9abe_ac01_eb1a_c46607435850 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

fastapi/concurrency.py lines 18–41

async def contextmanager_in_threadpool(
    cm: AbstractContextManager[_T],
) -> AsyncGenerator[_T, None]:
    # blocking __exit__ from running waiting on a free thread
    # can create race conditions/deadlocks if the context manager itself
    # has its own internal pool (e.g. a database connection pool)
    # to avoid this we let __exit__ run without a capacity limit
    # since we're creating a new limiter for each call, any non-zero limit
    # works (1 is arbitrary)
    exit_limiter = CapacityLimiter(1)
    try:
        yield await run_in_threadpool(cm.__enter__)
    except Exception as e:
        ok = bool(
            await anyio.to_thread.run_sync(
                cm.__exit__, type(e), e, e.__traceback__, limiter=exit_limiter
            )
        )
        if not ok:
            raise e
    else:
        await anyio.to_thread.run_sync(
            cm.__exit__, None, None, None, limiter=exit_limiter
        )