_UsesScopesCache = dict[int, tuple[Dependant, bool]]
+_CALLABLE_CLASSIFICATION_CACHE_SIZE = 4096
class _CallIdentity:
return [dep for dep in dependant.dependencies if _is_security_scheme(dependant=dep)]
-@lru_cache(maxsize=1024)
+@lru_cache(maxsize=_CALLABLE_CLASSIFICATION_CACHE_SIZE)
def _is_gen_callable_cached(call_identity: _CallIdentity) -> bool:
call = call_identity.call
if inspect.isgeneratorfunction(_impartial(call)) or inspect.isgeneratorfunction(
return _is_gen_callable_cached(_CallIdentity(call))
-@lru_cache(maxsize=1024)
+@lru_cache(maxsize=_CALLABLE_CLASSIFICATION_CACHE_SIZE)
def _is_async_gen_callable_cached(call_identity: _CallIdentity) -> bool:
call = call_identity.call
if inspect.isasyncgenfunction(_impartial(call)) or inspect.isasyncgenfunction(
return _is_async_gen_callable_cached(_CallIdentity(call))
-@lru_cache(maxsize=1024)
+@lru_cache(maxsize=_CALLABLE_CLASSIFICATION_CACHE_SIZE)
def _is_coroutine_callable_cached(call_identity: _CallIdentity) -> bool:
call = call_identity.call
if inspect.isroutine(_impartial(call)) and iscoroutinefunction(_impartial(call)):
-from collections.abc import AsyncGenerator, Generator
+from collections.abc import AsyncGenerator, Callable, Generator
from typing import Any
from fastapi.dependencies.models import (
cache_info = cached_function.cache_info()
assert cache_info.hits == 1
assert cache_info.misses == 1
- assert cache_info.maxsize == 1024
+ assert cache_info.maxsize == 4096
+
+
+def test_callable_classification_cache_supports_large_apps() -> None:
+ callables: list[Callable[[], None]] = [lambda: None for _ in range(3000)]
+
+ for classifier, cached_classifier in (
+ (_is_gen_callable, _is_gen_callable_cached),
+ (_is_async_gen_callable, _is_async_gen_callable_cached),
+ (_is_coroutine_callable, _is_coroutine_callable_cached),
+ ):
+ cached_classifier.cache_clear()
+
+ for _ in range(2):
+ assert all(not classifier(call) for call in callables)
+
+ cache_info = cached_classifier.cache_info()
+ assert cache_info.hits == len(callables)
+ assert cache_info.misses == len(callables)
+ assert cache_info.maxsize == 4096
+ cached_classifier.cache_clear()
def test_unhashable_callable_classification() -> None: