concurrency.py — fastapi Source File

Architecture documentation for concurrency.py, a python file in the fastapi codebase. 6 imports, 2 dependents.

File python FastAPI Routing 6 imports 2 dependents 1 functions

Entity Profile

FastAPI→ Routing→ concurrency.py — fastapi Source File

Dependency Diagram

graph LR
  e587e889_5c93_286d_8323_600fa8470a5c["concurrency.py"]
  07d79a2e_d4e9_0bbb_be90_936274444c8c["collections.abc"]
  e587e889_5c93_286d_8323_600fa8470a5c --> 07d79a2e_d4e9_0bbb_be90_936274444c8c
  a88eb002_9197_73f3_410d_ee5315767f34["contextlib"]
  e587e889_5c93_286d_8323_600fa8470a5c --> a88eb002_9197_73f3_410d_ee5315767f34
  0dda2280_3359_8460_301c_e98c77e78185["typing"]
  e587e889_5c93_286d_8323_600fa8470a5c --> 0dda2280_3359_8460_301c_e98c77e78185
  230d018b_3e12_b895_a647_4b018f333aaf["anyio.to_thread"]
  e587e889_5c93_286d_8323_600fa8470a5c --> 230d018b_3e12_b895_a647_4b018f333aaf
  c56b4bb1_311b_6874_0f07_e8e8c69d3e0b["anyio"]
  e587e889_5c93_286d_8323_600fa8470a5c --> c56b4bb1_311b_6874_0f07_e8e8c69d3e0b
  9d13606e_2eae_0bbe_bc8a_e5637b0e58e4["starlette.concurrency"]
  e587e889_5c93_286d_8323_600fa8470a5c --> 9d13606e_2eae_0bbe_bc8a_e5637b0e58e4
  9e602cbf_3139_86ae_5666_97b8806942de["utils.py"]
  9e602cbf_3139_86ae_5666_97b8806942de --> e587e889_5c93_286d_8323_600fa8470a5c
  23108d30_2765_dee1_a01d_c92aecd86c64["test_dependency_wrapped.py"]
  23108d30_2765_dee1_a01d_c92aecd86c64 --> e587e889_5c93_286d_8323_600fa8470a5c
  style e587e889_5c93_286d_8323_600fa8470a5c fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

from collections.abc import AsyncGenerator
from contextlib import AbstractContextManager
from contextlib import asynccontextmanager as asynccontextmanager
from typing import TypeVar

import anyio.to_thread
from anyio import CapacityLimiter
from starlette.concurrency import iterate_in_threadpool as iterate_in_threadpool  # noqa
from starlette.concurrency import run_in_threadpool as run_in_threadpool  # noqa
from starlette.concurrency import (  # noqa
    run_until_first_complete as run_until_first_complete,
)

_T = TypeVar("_T")


@asynccontextmanager
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
        )