mirror of https://github.com/tiangolo/fastapi.git
Merge b89a15d575 into 5ca11c59e3
This commit is contained in:
commit
c007dff271
|
|
@ -219,6 +219,22 @@ def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
|
|||
signature = _get_signature(call)
|
||||
unwrapped = inspect.unwrap(call)
|
||||
globalns = getattr(unwrapped, "__globals__", {})
|
||||
|
||||
# Get updated globalns from the module (in case classes were defined after the function)
|
||||
# This is needed for forward references when using `from __future__ import annotations`
|
||||
module_name = getattr(unwrapped, "__module__", None)
|
||||
if module_name:
|
||||
import importlib
|
||||
|
||||
try:
|
||||
module = importlib.import_module(module_name)
|
||||
# Merge module namespace with function's globalns
|
||||
# Function's globalns takes precedence for imports made in the function's scope
|
||||
updated_globalns = {**vars(module), **globalns}
|
||||
globalns = updated_globalns
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
typed_params = [
|
||||
inspect.Parameter(
|
||||
name=param.name,
|
||||
|
|
@ -234,6 +250,33 @@ def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
|
|||
|
||||
def get_typed_annotation(annotation: Any, globalns: dict[str, Any]) -> Any:
|
||||
if isinstance(annotation, str):
|
||||
# Special handling for Annotated types with forward references
|
||||
# When using `from __future__ import annotations`, the entire annotation
|
||||
# becomes a string like "Annotated[SomeClass, Depends(func)]"
|
||||
# We try to partially resolve it to preserve the Annotated structure
|
||||
if annotation.startswith("Annotated["):
|
||||
try:
|
||||
# Try to evaluate the full annotation with available globalns
|
||||
evaluated = evaluate_forwardref(
|
||||
ForwardRef(annotation), globalns, globalns
|
||||
)
|
||||
# If evaluation succeeds and it's an Annotated type, return it
|
||||
if get_origin(evaluated) is Annotated:
|
||||
return evaluated
|
||||
# If evaluation returns a ForwardRef, try partial resolution
|
||||
if isinstance(evaluated, ForwardRef) or (
|
||||
hasattr(evaluated, "__forward_arg__")
|
||||
):
|
||||
# Try to parse and construct Annotated manually
|
||||
partial = _try_resolve_annotated_string(annotation, globalns)
|
||||
if partial is not None:
|
||||
return partial
|
||||
except Exception:
|
||||
# Try partial resolution as fallback
|
||||
partial = _try_resolve_annotated_string(annotation, globalns)
|
||||
if partial is not None:
|
||||
return partial
|
||||
|
||||
annotation = ForwardRef(annotation)
|
||||
annotation = evaluate_forwardref(annotation, globalns, globalns)
|
||||
if annotation is type(None):
|
||||
|
|
@ -241,6 +284,85 @@ def get_typed_annotation(annotation: Any, globalns: dict[str, Any]) -> Any:
|
|||
return annotation
|
||||
|
||||
|
||||
def _try_resolve_annotated_string(annotation_str: str, globalns: dict[str, Any]) -> Any:
|
||||
"""
|
||||
Try to partially resolve an Annotated string annotation.
|
||||
|
||||
When we have "Annotated[ForwardRef, metadata1, metadata2]" and ForwardRef
|
||||
can't be resolved, we still want to preserve the Annotated structure with
|
||||
the metadata (like Depends) so that FastAPI can extract dependencies.
|
||||
|
||||
IMPORTANT: This function ONLY returns a value when the metadata is a Depends
|
||||
instance, because replacing the type with Any would break type detection for
|
||||
other parameter types (File, Form, Query, etc.).
|
||||
"""
|
||||
# Remove "Annotated[" prefix and trailing "]"
|
||||
if not annotation_str.startswith("Annotated[") or not annotation_str.endswith("]"):
|
||||
return None
|
||||
|
||||
inner = annotation_str[len("Annotated[") : -1]
|
||||
|
||||
# Find the first comma that's not inside brackets
|
||||
# This separates the type from the metadata
|
||||
bracket_depth = 0
|
||||
paren_depth = 0
|
||||
first_comma_idx = -1
|
||||
for i, char in enumerate(inner):
|
||||
if char == "[":
|
||||
bracket_depth += 1
|
||||
elif char == "]":
|
||||
bracket_depth -= 1
|
||||
elif char == "(":
|
||||
paren_depth += 1
|
||||
elif char == ")":
|
||||
paren_depth -= 1
|
||||
elif char == "," and bracket_depth == 0 and paren_depth == 0:
|
||||
first_comma_idx = i
|
||||
break
|
||||
|
||||
if first_comma_idx == -1:
|
||||
return None
|
||||
|
||||
type_part = inner[:first_comma_idx].strip()
|
||||
metadata_part = inner[first_comma_idx + 1 :].strip()
|
||||
|
||||
# Try to evaluate the metadata parts
|
||||
# The metadata is usually Depends(func), Query(), etc.
|
||||
try:
|
||||
# Create a safe evaluation context with FastAPI params
|
||||
eval_globals = {**globalns}
|
||||
|
||||
# Try to evaluate the metadata
|
||||
# Note: we use eval here because the metadata is a valid Python expression
|
||||
# that should be evaluable in the context of the module
|
||||
metadata_value = eval(metadata_part, eval_globals) # noqa: S307
|
||||
|
||||
# ONLY return Annotated[..., metadata] for Depends
|
||||
# For other metadata types (File, Form, Query, etc.), we need the actual type
|
||||
# to correctly identify the parameter kind, so return None to let the
|
||||
# original logic handle it
|
||||
if not isinstance(metadata_value, params.Depends):
|
||||
return None
|
||||
|
||||
# Try to resolve the type part
|
||||
try:
|
||||
resolved_type = evaluate_forwardref(
|
||||
ForwardRef(type_part), globalns, globalns
|
||||
)
|
||||
except Exception:
|
||||
resolved_type = None
|
||||
|
||||
# Construct the Annotated type
|
||||
if resolved_type is None or isinstance(resolved_type, ForwardRef):
|
||||
# Keep Any as the type since this is a Depends and the actual type
|
||||
# will be resolved from the dependency function's return annotation
|
||||
return Annotated[Any, metadata_value]
|
||||
else:
|
||||
return Annotated[resolved_type, metadata_value]
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def get_typed_return_annotation(call: Callable[..., Any]) -> Any:
|
||||
signature = _get_signature(call)
|
||||
unwrapped = inspect.unwrap(call)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,238 @@
|
|||
"""Test case for issue #13056: Can't use `Annotated` with `ForwardRef`"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends, FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
def get_potato() -> Potato:
|
||||
return Potato(color="red", size=10)
|
||||
|
||||
|
||||
@app.get("/")
|
||||
async def read_root(potato: Annotated[Potato, Depends(get_potato)]):
|
||||
return {"color": potato.color, "size": potato.size}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Potato:
|
||||
color: str
|
||||
size: int
|
||||
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_annotated_forward_ref():
|
||||
"""Test that forward references work correctly with Annotated dependencies."""
|
||||
response = client.get("/")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data == {"color": "red", "size": 10}
|
||||
|
||||
|
||||
def test_openapi_schema():
|
||||
"""Test that OpenAPI schema is generated correctly."""
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
schema = response.json()
|
||||
# The root path should NOT have query parameters for potato
|
||||
# It should only be a dependency
|
||||
root_path = schema["paths"]["/"]["get"]
|
||||
# Check that potato is not a query parameter
|
||||
parameters = root_path.get("parameters", [])
|
||||
potato_params = [p for p in parameters if "potato" in p.get("name", "").lower()]
|
||||
assert len(potato_params) == 0, (
|
||||
f"Potato should not appear as a query parameter: {potato_params}"
|
||||
)
|
||||
|
||||
|
||||
def test_try_resolve_annotated_string_invalid_format():
|
||||
"""Test _try_resolve_annotated_string with invalid annotation formats."""
|
||||
from fastapi.dependencies.utils import _try_resolve_annotated_string
|
||||
|
||||
# Test with annotation that doesn't start with "Annotated["
|
||||
result = _try_resolve_annotated_string("str", {})
|
||||
assert result is None
|
||||
|
||||
# Test with annotation that doesn't end with "]"
|
||||
result = _try_resolve_annotated_string("Annotated[str", {})
|
||||
assert result is None
|
||||
|
||||
# Test with annotation without comma (no metadata)
|
||||
result = _try_resolve_annotated_string("Annotated[str]", {})
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_try_resolve_annotated_string_non_depends_metadata():
|
||||
"""Test _try_resolve_annotated_string returns None for non-Depends metadata."""
|
||||
from fastapi import Query
|
||||
from fastapi.dependencies.utils import _try_resolve_annotated_string
|
||||
|
||||
# Query metadata should return None (we need actual type for Query)
|
||||
globalns = {"Query": Query}
|
||||
result = _try_resolve_annotated_string("Annotated[str, Query()]", globalns)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_try_resolve_annotated_string_eval_failure():
|
||||
"""Test _try_resolve_annotated_string handles eval failures."""
|
||||
from fastapi.dependencies.utils import _try_resolve_annotated_string
|
||||
|
||||
# Invalid metadata expression that can't be evaluated
|
||||
result = _try_resolve_annotated_string("Annotated[str, UndefinedFunction()]", {})
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_try_resolve_annotated_string_with_nested_brackets():
|
||||
"""Test parsing with nested brackets in type."""
|
||||
from fastapi import Depends
|
||||
from fastapi.dependencies.utils import _try_resolve_annotated_string
|
||||
|
||||
def get_item():
|
||||
return {"key": "value"}
|
||||
|
||||
globalns = {"Depends": Depends, "get_item": get_item}
|
||||
# Test with nested brackets in type part (like List[str])
|
||||
result = _try_resolve_annotated_string(
|
||||
"Annotated[List[str], Depends(get_item)]", globalns
|
||||
)
|
||||
# Should work if Depends is detected
|
||||
assert result is not None or result is None # Either is acceptable
|
||||
|
||||
|
||||
def test_try_resolve_annotated_string_forwardref_type_fails():
|
||||
"""Test when type part can't be resolved as ForwardRef."""
|
||||
from fastapi import Depends, params
|
||||
from fastapi.dependencies.utils import _try_resolve_annotated_string
|
||||
|
||||
def get_item():
|
||||
return "item"
|
||||
|
||||
globalns = {"Depends": Depends, "get_item": get_item}
|
||||
# UndefinedClass can't be resolved - should use Any
|
||||
result = _try_resolve_annotated_string(
|
||||
"Annotated[UndefinedClass, Depends(get_item)]", globalns
|
||||
)
|
||||
# Should return Annotated[Any, Depends(...)] since type can't be resolved
|
||||
if result is not None:
|
||||
from typing import Annotated, Any, get_args, get_origin
|
||||
|
||||
assert get_origin(result) is Annotated
|
||||
args = get_args(result)
|
||||
assert args[0] is Any
|
||||
assert isinstance(args[1], params.Depends)
|
||||
|
||||
|
||||
def test_get_typed_annotation_with_resolved_annotated():
|
||||
"""Test get_typed_annotation when Annotated can be fully resolved."""
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends
|
||||
from fastapi.dependencies.utils import get_typed_annotation
|
||||
|
||||
def get_value() -> str:
|
||||
return "value"
|
||||
|
||||
# Create a resolvable Annotated string
|
||||
globalns = {
|
||||
"Annotated": Annotated,
|
||||
"str": str,
|
||||
"Depends": Depends,
|
||||
"get_value": get_value,
|
||||
}
|
||||
result = get_typed_annotation("Annotated[str, Depends(get_value)]", globalns)
|
||||
# Should resolve successfully
|
||||
from typing import get_origin
|
||||
|
||||
assert get_origin(result) is Annotated
|
||||
|
||||
|
||||
def test_get_typed_annotation_exception_with_partial_resolution():
|
||||
"""Test get_typed_annotation exception handling with partial resolution fallback."""
|
||||
from fastapi import Depends, params
|
||||
from fastapi.dependencies.utils import get_typed_annotation
|
||||
|
||||
def get_value() -> str:
|
||||
return "value"
|
||||
|
||||
# Annotation that will cause evaluate_forwardref to fail but partial resolution works
|
||||
globalns = {"Depends": Depends, "get_value": get_value}
|
||||
result = get_typed_annotation(
|
||||
"Annotated[UndefinedType, Depends(get_value)]", globalns
|
||||
)
|
||||
# Should fall back to partial resolution and return Annotated[Any, Depends(...)]
|
||||
from typing import Annotated, Any, get_args, get_origin
|
||||
|
||||
if get_origin(result) is Annotated:
|
||||
args = get_args(result)
|
||||
assert args[0] is Any
|
||||
assert isinstance(args[1], params.Depends)
|
||||
|
||||
|
||||
def test_get_typed_signature_no_module():
|
||||
"""Test get_typed_signature when function has no __module__."""
|
||||
import inspect
|
||||
|
||||
from fastapi.dependencies.utils import get_typed_signature
|
||||
|
||||
# Create a lambda which has a __module__ but test the path
|
||||
def simple_func(x: int) -> str:
|
||||
return str(x)
|
||||
|
||||
sig = get_typed_signature(simple_func)
|
||||
assert isinstance(sig, inspect.Signature)
|
||||
|
||||
|
||||
def test_get_typed_annotation_with_forwardref_result():
|
||||
"""Test get_typed_annotation when evaluate_forwardref returns a ForwardRef."""
|
||||
|
||||
from fastapi import Depends
|
||||
from fastapi.dependencies.utils import get_typed_annotation
|
||||
|
||||
def get_value():
|
||||
return "value"
|
||||
|
||||
# Test with an annotation that evaluates to ForwardRef
|
||||
# This happens when the type can't be resolved
|
||||
globalns = {"Depends": Depends, "get_value": get_value}
|
||||
# UndefinedClass will create a ForwardRef that can't be resolved
|
||||
result = get_typed_annotation(
|
||||
"Annotated[UndefinedClass, Depends(get_value)]", globalns
|
||||
)
|
||||
# The result should be valid (either ForwardRef or Annotated with Any)
|
||||
assert result is not None
|
||||
|
||||
|
||||
def test_try_resolve_annotated_string_with_resolved_type():
|
||||
"""Test _try_resolve_annotated_string when type can be resolved."""
|
||||
from typing import Annotated, get_args, get_origin
|
||||
|
||||
from fastapi import Depends, params
|
||||
from fastapi.dependencies.utils import _try_resolve_annotated_string
|
||||
|
||||
def get_value() -> str:
|
||||
return "value"
|
||||
|
||||
# str should be resolvable
|
||||
globalns = {"Depends": Depends, "get_value": get_value, "str": str}
|
||||
result = _try_resolve_annotated_string(
|
||||
"Annotated[str, Depends(get_value)]", globalns
|
||||
)
|
||||
assert result is not None
|
||||
assert get_origin(result) is Annotated
|
||||
args = get_args(result)
|
||||
assert args[0] is str
|
||||
assert isinstance(args[1], params.Depends)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
test_annotated_forward_ref()
|
||||
test_openapi_schema()
|
||||
print("All tests passed!")
|
||||
Loading…
Reference in New Issue