Fix _endpoint_context_cache using id() which risks stale entries after GC

The _endpoint_context_cache in routing.py used id(func) as dict keys to
cache endpoint context (source file, line number, function name) for
error messages. This has two problems:

1. ID reuse after garbage collection. Python id() returns the memory
   address of an object. Once an object is garbage collected, its ID can
   be reassigned to a newly created object. If an endpoint function were
   deallocated and a new function reused the same ID, the cache would
   return stale context (wrong file, line, function name) for error
   messages.

2. No eviction. The module-level dict grows unboundedly for every unique
   endpoint function. While each entry is small (~3 strings), this
   matters in scenarios where FastAPI instances are dynamically created
   and destroyed in a single process (e.g., test suites, multi-tenant
   embedding).

Fix: key the cache on the function object itself instead of id(func).
Since Python functions use identity-based hashing by default, lookup
performance is identical. The dict now holds a strong reference to each
function, which means the function cannot be garbage collected while its
cache entry exists, and therefore its ID cannot be reused for a different
object. In practice, endpoint functions are already held alive by the
router for the app lifetime, so this does not change object lifetimes.
This commit is contained in:
Charisn 2026-02-15 13:08:44 +02:00
parent ed12105cce
commit 259bd33625
1 changed files with 7 additions and 7 deletions

View File

@ -232,16 +232,16 @@ class _DefaultLifespan:
return self
# Cache for endpoint context to avoid re-extracting on every request
_endpoint_context_cache: dict[int, EndpointContext] = {}
# Cache for endpoint context to avoid re-extracting on every request.
# Keyed on the function object itself (not id()) so that entries remain valid
# even if a function is garbage collected and its id is reused by a new object.
_endpoint_context_cache: dict[Callable[..., Any], EndpointContext] = {}
def _extract_endpoint_context(func: Any) -> EndpointContext:
"""Extract endpoint context with caching to avoid repeated file I/O."""
func_id = id(func)
if func_id in _endpoint_context_cache:
return _endpoint_context_cache[func_id]
if func in _endpoint_context_cache:
return _endpoint_context_cache[func]
try:
ctx: EndpointContext = {}
@ -255,7 +255,7 @@ def _extract_endpoint_context(func: Any) -> EndpointContext:
except Exception:
ctx = EndpointContext()
_endpoint_context_cache[func_id] = ctx
_endpoint_context_cache[func] = ctx
return ctx