Browse Source

Add docstrings to internal utility functions

Add docstrings to contextmanager_in_threadpool and
generate_encoders_by_class_tuples to document their purpose and
behavior for contributors navigating the codebase.

Co-Authored-By: Claude Opus 4.6 <[email protected]>
pull/15018/head
Anandesh Sharma 5 months ago
parent
commit
4e96b92397
  1. 15
      fastapi/concurrency.py
  2. 6
      fastapi/encoders.py

15
fastapi/concurrency.py

@ -18,12 +18,15 @@ _T = TypeVar("_T")
async def contextmanager_in_threadpool( async def contextmanager_in_threadpool(
cm: AbstractContextManager[_T], cm: AbstractContextManager[_T],
) -> AsyncGenerator[_T, None]: ) -> AsyncGenerator[_T, None]:
# blocking __exit__ from running waiting on a free thread """Run a synchronous context manager in a thread pool.
# can create race conditions/deadlocks if the context manager itself
# has its own internal pool (e.g. a database connection pool) Enters and exits the context manager in a worker thread so that
# to avoid this we let __exit__ run without a capacity limit blocking ``__enter__`` / ``__exit__`` calls do not block the async
# since we're creating a new limiter for each call, any non-zero limit event loop. The ``__exit__`` call uses its own
# works (1 is arbitrary) :class:`~anyio.CapacityLimiter` to avoid deadlocks when the context
manager holds resources from a bounded pool (e.g. database
connections).
"""
exit_limiter = CapacityLimiter(1) exit_limiter = CapacityLimiter(1)
try: try:
yield await run_in_threadpool(cm.__enter__) yield await run_in_threadpool(cm.__enter__)

6
fastapi/encoders.py

@ -98,6 +98,12 @@ ENCODERS_BY_TYPE: dict[type[Any], Callable[[Any], Any]] = {
def generate_encoders_by_class_tuples( def generate_encoders_by_class_tuples(
type_encoder_map: dict[Any, Callable[[Any], Any]], type_encoder_map: dict[Any, Callable[[Any], Any]],
) -> dict[Callable[[Any], Any], tuple[Any, ...]]: ) -> dict[Callable[[Any], Any], tuple[Any, ...]]:
"""Invert a ``{type: encoder}`` mapping to ``{encoder: (types...)}``.
This allows :func:`jsonable_encoder` to perform efficient
``isinstance`` checks against all types that share the same encoder
function in a single call.
"""
encoders_by_class_tuples: dict[Callable[[Any], Any], tuple[Any, ...]] = defaultdict( encoders_by_class_tuples: dict[Callable[[Any], Any], tuple[Any, ...]] = defaultdict(
tuple tuple
) )

Loading…
Cancel
Save