Compare commits

...

22 Commits

Author SHA1 Message Date
Sebastián Ramírez
b5ca13249e 🔖 Release version 0.124.0 2025-12-06 14:09:51 +01:00
github-actions[bot]
a2cef707e3 📝 Update release notes
[skip ci]
2025-12-06 12:23:23 +00:00
Yuji Teshima
5b6245666b ✏️ Fix typo in scripts/mkdocs_hooks.py (#14457) 2025-12-06 13:23:01 +01:00
github-actions[bot]
dbd34f1578 📝 Update release notes
[skip ci]
2025-12-06 12:22:24 +00:00
Savannah Ostrowski
e1117f7550 🚸 Improve tracebacks by adding endpoint metadata (#14306)
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
2025-12-06 12:21:57 +00:00
Sebastián Ramírez
08b09e5236 🔖 Release version 0.123.10 2025-12-05 22:26:36 +01:00
github-actions[bot]
e7d7038dfa 📝 Update release notes
[skip ci]
2025-12-05 21:21:29 +00:00
Motov Yurii
da0ffab0b2 🐛 Fix using class (not instance) dependency that has __call__ method (#14458)
Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
2025-12-05 21:21:05 +00:00
github-actions[bot]
516169428d 📝 Update release notes
[skip ci]
2025-12-05 20:19:54 +00:00
Motov Yurii
812a1926f0 🐛 Fix separate_input_output_schemas=False with computed_field (#14453) 2025-12-05 21:19:30 +01:00
Sebastián Ramírez
f0dd1046a6 🔖 Release version 0.123.9 2025-12-04 23:23:21 +01:00
github-actions[bot]
188d631011 📝 Update release notes
[skip ci]
2025-12-04 22:22:25 +00:00
Sebastián Ramírez
0b5fa563cd 🐛 Fix OAuth2 scopes in OpenAPI in extra corner cases, parent dependency with scopes, sub-dependency security scheme without scopes (#14459) 2025-12-04 23:22:01 +01:00
Sebastián Ramírez
eb1d50479b 🔖 Release version 0.123.8 2025-12-04 14:01:00 +01:00
github-actions[bot]
e248a4d22b 📝 Update release notes
[skip ci]
2025-12-04 12:59:45 +00:00
Sebastián Ramírez
0ec4bafca2 🐛 Fix OpenAPI security scheme OAuth2 scopes declaration, deduplicate security schemes with different scopes (#14455) 2025-12-04 13:59:24 +01:00
Sebastián Ramírez
603df6e36f 🔖 Release version 0.123.7 2025-12-04 09:27:38 +01:00
github-actions[bot]
6c565482cf 📝 Update release notes
[skip ci]
2025-12-04 08:18:55 +00:00
chaen
861598b4e3 🐛 Fix evaluating stringified annotations in Python 3.10 (#11355)
Co-authored-by: Sofie Van Landeghem <svlandeg@users.noreply.github.com>
Co-authored-by: svlandeg <svlandeg@github.com>
Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
2025-12-04 09:18:32 +01:00
Sebastián Ramírez
811fa89875 🔖 Release version 0.123.6 2025-12-04 08:33:11 +01:00
github-actions[bot]
6c6b9d7a2b 📝 Update release notes
[skip ci]
2025-12-04 07:29:53 +00:00
Sebastián Ramírez
bba4d4c95e 🐛 Fix support for functools wraps and partial combined, for async and regular functions and classes in path operations and dependencies (#14448)
Co-authored-by: Yurii Motov <yurii.motov.monte@gmail.com>
2025-12-04 08:29:28 +01:00
16 changed files with 1359 additions and 78 deletions

View File

@@ -7,6 +7,47 @@ hide:
## Latest Changes
## 0.124.0
### Features
* 🚸 Improve tracebacks by adding endpoint metadata. PR [#14306](https://github.com/fastapi/fastapi/pull/14306) by [@savannahostrowski](https://github.com/savannahostrowski).
### Internal
* ✏️ Fix typo in `scripts/mkdocs_hooks.py`. PR [#14457](https://github.com/fastapi/fastapi/pull/14457) by [@yujiteshima](https://github.com/yujiteshima).
## 0.123.10
### Fixes
* 🐛 Fix using class (not instance) dependency that has `__call__` method. PR [#14458](https://github.com/fastapi/fastapi/pull/14458) by [@YuriiMotov](https://github.com/YuriiMotov).
* 🐛 Fix `separate_input_output_schemas=False` with `computed_field`. PR [#14453](https://github.com/fastapi/fastapi/pull/14453) by [@YuriiMotov](https://github.com/YuriiMotov).
## 0.123.9
### Fixes
* 🐛 Fix OAuth2 scopes in OpenAPI in extra corner cases, parent dependency with scopes, sub-dependency security scheme without scopes. PR [#14459](https://github.com/fastapi/fastapi/pull/14459) by [@tiangolo](https://github.com/tiangolo).
## 0.123.8
### Fixes
* 🐛 Fix OpenAPI security scheme OAuth2 scopes declaration, deduplicate security schemes with different scopes. PR [#14455](https://github.com/fastapi/fastapi/pull/14455) by [@tiangolo](https://github.com/tiangolo).
## 0.123.7
### Fixes
* 🐛 Fix evaluating stringified annotations in Python 3.10. PR [#11355](https://github.com/fastapi/fastapi/pull/11355) by [@chaen](https://github.com/chaen).
## 0.123.6
### Fixes
* 🐛 Fix support for functools wraps and partial combined, for async and regular functions and classes in path operations and dependencies. PR [#14448](https://github.com/fastapi/fastapi/pull/14448) by [@tiangolo](https://github.com/tiangolo).
## 0.123.5
### Features

View File

@@ -1,6 +1,6 @@
"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
__version__ = "0.123.5"
__version__ = "0.124.0"
from starlette import status as status

View File

@@ -171,6 +171,13 @@ def _get_model_config(model: BaseModel) -> Any:
return model.model_config
def _has_computed_fields(field: ModelField) -> bool:
computed_fields = field._type_adapter.core_schema.get("schema", {}).get(
"computed_fields", []
)
return len(computed_fields) > 0
def get_schema_from_model_field(
*,
field: ModelField,
@@ -180,12 +187,9 @@ def get_schema_from_model_field(
],
separate_input_output_schemas: bool = True,
) -> Dict[str, Any]:
computed_fields = field._type_adapter.core_schema.get("schema", {}).get(
"computed_fields", []
)
override_mode: Union[Literal["validation"], None] = (
None
if (separate_input_output_schemas or len(computed_fields) > 0)
if (separate_input_output_schemas or _has_computed_fields(field))
else "validation"
)
# This expects that GenerateJsonSchema was already used to generate the definitions
@@ -208,15 +212,7 @@ def get_definitions(
Dict[Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue],
Dict[str, Dict[str, Any]],
]:
has_computed_fields: bool = any(
field._type_adapter.core_schema.get("schema", {}).get("computed_fields", [])
for field in fields
)
schema_generator = GenerateJsonSchema(ref_template=REF_TEMPLATE)
override_mode: Union[Literal["validation"], None] = (
None if (separate_input_output_schemas or has_computed_fields) else "validation"
)
validation_fields = [field for field in fields if field.mode == "validation"]
serialization_fields = [field for field in fields if field.mode == "serialization"]
flat_validation_models = get_flat_models_from_fields(
@@ -246,9 +242,16 @@ def get_definitions(
unique_flat_model_fields = {
f for f in flat_model_fields if f.type_ not in input_types
}
inputs = [
(field, override_mode or field.mode, field._type_adapter.core_schema)
(
field,
(
field.mode
if (separate_input_output_schemas or _has_computed_fields(field))
else "validation"
),
field._type_adapter.core_schema,
)
for field in list(fields) + list(unique_flat_model_fields)
]
field_mapping, definitions = schema_generator.generate_definitions(inputs=inputs)

View File

@@ -2,7 +2,7 @@ import inspect
import sys
from dataclasses import dataclass, field
from functools import cached_property, partial
from typing import Any, Callable, List, Optional, Sequence, Union
from typing import Any, Callable, List, Optional, Union
from fastapi._compat import ModelField
from fastapi.security.base import SecurityBase
@@ -15,10 +15,17 @@ else: # pragma: no cover
from asyncio import iscoroutinefunction
@dataclass
class SecurityRequirement:
security_scheme: SecurityBase
scopes: Optional[Sequence[str]] = None
def _unwrapped_call(call: Optional[Callable[..., Any]]) -> Any:
if call is None:
return call # pragma: no cover
unwrapped = inspect.unwrap(_impartial(call))
return unwrapped
def _impartial(func: Callable[..., Any]) -> Callable[..., Any]:
while isinstance(func, partial):
func = func.func
return func
@dataclass
@@ -29,7 +36,6 @@ class Dependant:
cookie_params: List[ModelField] = field(default_factory=list)
body_params: List[ModelField] = field(default_factory=list)
dependencies: List["Dependant"] = field(default_factory=list)
security_requirements: List[SecurityRequirement] = field(default_factory=list)
name: Optional[str] = None
call: Optional[Callable[..., Any]] = None
request_param_name: Optional[str] = None
@@ -70,42 +76,113 @@ class Dependant:
return True
if self.security_scopes_param_name is not None:
return True
if self._is_security_scheme:
return True
for sub_dep in self.dependencies:
if sub_dep._uses_scopes:
return True
return False
@cached_property
def _unwrapped_call(self) -> Any:
def _is_security_scheme(self) -> bool:
if self.call is None:
return self.call # pragma: no cover
unwrapped = inspect.unwrap(self.call)
if isinstance(unwrapped, partial):
unwrapped = unwrapped.func
return False # pragma: no cover
unwrapped = _unwrapped_call(self.call)
return isinstance(unwrapped, SecurityBase)
# Mainly to get the type of SecurityBase, but it's the same self.call
@cached_property
def _security_scheme(self) -> SecurityBase:
unwrapped = _unwrapped_call(self.call)
assert isinstance(unwrapped, SecurityBase)
return unwrapped
@cached_property
def _security_dependencies(self) -> List["Dependant"]:
security_deps = [dep for dep in self.dependencies if dep._is_security_scheme]
return security_deps
@cached_property
def is_gen_callable(self) -> bool:
if inspect.isgeneratorfunction(self._unwrapped_call):
if self.call is None:
return False # pragma: no cover
if inspect.isgeneratorfunction(
_impartial(self.call)
) or inspect.isgeneratorfunction(_unwrapped_call(self.call)):
return True
dunder_call = getattr(self._unwrapped_call, "__call__", None) # noqa: B004
return inspect.isgeneratorfunction(dunder_call)
if inspect.isclass(_unwrapped_call(self.call)):
return False
dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
if dunder_call is None:
return False # pragma: no cover
if inspect.isgeneratorfunction(
_impartial(dunder_call)
) or inspect.isgeneratorfunction(_unwrapped_call(dunder_call)):
return True
dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
if dunder_unwrapped_call is None:
return False # pragma: no cover
if inspect.isgeneratorfunction(
_impartial(dunder_unwrapped_call)
) or inspect.isgeneratorfunction(_unwrapped_call(dunder_unwrapped_call)):
return True
return False
@cached_property
def is_async_gen_callable(self) -> bool:
if inspect.isasyncgenfunction(self._unwrapped_call):
if self.call is None:
return False # pragma: no cover
if inspect.isasyncgenfunction(
_impartial(self.call)
) or inspect.isasyncgenfunction(_unwrapped_call(self.call)):
return True
dunder_call = getattr(self._unwrapped_call, "__call__", None) # noqa: B004
return inspect.isasyncgenfunction(dunder_call)
if inspect.isclass(_unwrapped_call(self.call)):
return False
dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
if dunder_call is None:
return False # pragma: no cover
if inspect.isasyncgenfunction(
_impartial(dunder_call)
) or inspect.isasyncgenfunction(_unwrapped_call(dunder_call)):
return True
dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
if dunder_unwrapped_call is None:
return False # pragma: no cover
if inspect.isasyncgenfunction(
_impartial(dunder_unwrapped_call)
) or inspect.isasyncgenfunction(_unwrapped_call(dunder_unwrapped_call)):
return True
return False
@cached_property
def is_coroutine_callable(self) -> bool:
if inspect.isroutine(self._unwrapped_call):
return iscoroutinefunction(self._unwrapped_call)
if inspect.isclass(self._unwrapped_call):
if self.call is None:
return False # pragma: no cover
if inspect.isroutine(_impartial(self.call)) and iscoroutinefunction(
_impartial(self.call)
):
return True
if inspect.isroutine(_unwrapped_call(self.call)) and iscoroutinefunction(
_unwrapped_call(self.call)
):
return True
if inspect.isclass(_unwrapped_call(self.call)):
return False
dunder_call = getattr(self._unwrapped_call, "__call__", None) # noqa: B004
return iscoroutinefunction(dunder_call)
dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
if dunder_call is None:
return False # pragma: no cover
if iscoroutinefunction(_impartial(dunder_call)) or iscoroutinefunction(
_unwrapped_call(dunder_call)
):
return True
dunder_unwrapped_call = getattr(_unwrapped_call(self.call), "__call__", None) # noqa: B004
if dunder_unwrapped_call is None:
return False # pragma: no cover
if iscoroutinefunction(
_impartial(dunder_unwrapped_call)
) or iscoroutinefunction(_unwrapped_call(dunder_unwrapped_call)):
return True
return False
@cached_property
def computed_scope(self) -> Union[str, None]:

View File

@@ -1,5 +1,6 @@
import dataclasses
import inspect
import sys
from contextlib import AsyncExitStack, contextmanager
from copy import copy, deepcopy
from dataclasses import dataclass
@@ -54,10 +55,9 @@ from fastapi.concurrency import (
asynccontextmanager,
contextmanager_in_threadpool,
)
from fastapi.dependencies.models import Dependant, SecurityRequirement
from fastapi.dependencies.models import Dependant
from fastapi.exceptions import DependencyScopeError
from fastapi.logger import logger
from fastapi.security.base import SecurityBase
from fastapi.security.oauth2 import SecurityScopes
from fastapi.types import DependencyCacheKey
from fastapi.utils import create_model_field, get_path_param_names
@@ -141,10 +141,14 @@ def get_flat_dependant(
*,
skip_repeats: bool = False,
visited: Optional[List[DependencyCacheKey]] = None,
parent_oauth_scopes: Optional[List[str]] = None,
) -> Dependant:
if visited is None:
visited = []
visited.append(dependant.cache_key)
use_parent_oauth_scopes = (parent_oauth_scopes or []) + (
dependant.oauth_scopes or []
)
flat_dependant = Dependant(
path_params=dependant.path_params.copy(),
@@ -152,22 +156,37 @@ def get_flat_dependant(
header_params=dependant.header_params.copy(),
cookie_params=dependant.cookie_params.copy(),
body_params=dependant.body_params.copy(),
security_requirements=dependant.security_requirements.copy(),
name=dependant.name,
call=dependant.call,
request_param_name=dependant.request_param_name,
websocket_param_name=dependant.websocket_param_name,
http_connection_param_name=dependant.http_connection_param_name,
response_param_name=dependant.response_param_name,
background_tasks_param_name=dependant.background_tasks_param_name,
security_scopes_param_name=dependant.security_scopes_param_name,
own_oauth_scopes=dependant.own_oauth_scopes,
parent_oauth_scopes=use_parent_oauth_scopes,
use_cache=dependant.use_cache,
path=dependant.path,
scope=dependant.scope,
)
for sub_dependant in dependant.dependencies:
if skip_repeats and sub_dependant.cache_key in visited:
continue
flat_sub = get_flat_dependant(
sub_dependant, skip_repeats=skip_repeats, visited=visited
sub_dependant,
skip_repeats=skip_repeats,
visited=visited,
parent_oauth_scopes=flat_dependant.oauth_scopes,
)
flat_dependant.dependencies.append(flat_sub)
flat_dependant.path_params.extend(flat_sub.path_params)
flat_dependant.query_params.extend(flat_sub.query_params)
flat_dependant.header_params.extend(flat_sub.header_params)
flat_dependant.cookie_params.extend(flat_sub.cookie_params)
flat_dependant.body_params.extend(flat_sub.body_params)
flat_dependant.security_requirements.extend(flat_sub.security_requirements)
flat_dependant.dependencies.extend(flat_sub.dependencies)
return flat_dependant
@@ -191,7 +210,10 @@ def get_flat_params(dependant: Dependant) -> List[ModelField]:
def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
signature = inspect.signature(call)
if sys.version_info >= (3, 10):
signature = inspect.signature(call, eval_str=True)
else:
signature = inspect.signature(call)
unwrapped = inspect.unwrap(call)
globalns = getattr(unwrapped, "__globals__", {})
typed_params = [
@@ -217,7 +239,10 @@ def get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any:
def get_typed_return_annotation(call: Callable[..., Any]) -> Any:
signature = inspect.signature(call)
if sys.version_info >= (3, 10):
signature = inspect.signature(call, eval_str=True)
else:
signature = inspect.signature(call)
unwrapped = inspect.unwrap(call)
annotation = signature.return_annotation
@@ -251,11 +276,6 @@ def get_dependant(
path_param_names = get_path_param_names(path)
endpoint_signature = get_typed_signature(call)
signature_params = endpoint_signature.parameters
if isinstance(call, SecurityBase):
security_requirement = SecurityRequirement(
security_scheme=call, scopes=current_scopes
)
dependant.security_requirements.append(security_requirement)
for param_name, param in signature_params.items():
is_path_param = param_name in path_param_names
param_details = analyze_param(
@@ -548,10 +568,10 @@ async def _solve_generator(
*, dependant: Dependant, stack: AsyncExitStack, sub_values: Dict[str, Any]
) -> Any:
assert dependant.call
if dependant.is_gen_callable:
cm = contextmanager_in_threadpool(contextmanager(dependant.call)(**sub_values))
elif dependant.is_async_gen_callable:
if dependant.is_async_gen_callable:
cm = asynccontextmanager(dependant.call)(**sub_values)
elif dependant.is_gen_callable:
cm = contextmanager_in_threadpool(contextmanager(dependant.call)(**sub_values))
return await stack.enter_async_context(cm)

View File

@@ -1,4 +1,4 @@
from typing import Any, Dict, Optional, Sequence, Type, Union
from typing import Any, Dict, Optional, Sequence, Type, TypedDict, Union
from annotated_doc import Doc
from pydantic import BaseModel, create_model
@@ -7,6 +7,13 @@ from starlette.exceptions import WebSocketException as StarletteWebSocketExcepti
from typing_extensions import Annotated
class EndpointContext(TypedDict, total=False):
function: str
path: str
file: str
line: int
class HTTPException(StarletteHTTPException):
"""
An HTTP exception you can raise in your own code to show errors to the client.
@@ -155,30 +162,72 @@ class DependencyScopeError(FastAPIError):
class ValidationException(Exception):
def __init__(self, errors: Sequence[Any]) -> None:
def __init__(
self,
errors: Sequence[Any],
*,
endpoint_ctx: Optional[EndpointContext] = None,
) -> None:
self._errors = errors
self.endpoint_ctx = endpoint_ctx
ctx = endpoint_ctx or {}
self.endpoint_function = ctx.get("function")
self.endpoint_path = ctx.get("path")
self.endpoint_file = ctx.get("file")
self.endpoint_line = ctx.get("line")
def errors(self) -> Sequence[Any]:
return self._errors
def _format_endpoint_context(self) -> str:
if not (self.endpoint_file and self.endpoint_line and self.endpoint_function):
if self.endpoint_path:
return f"\n Endpoint: {self.endpoint_path}"
return ""
context = f'\n File "{self.endpoint_file}", line {self.endpoint_line}, in {self.endpoint_function}'
if self.endpoint_path:
context += f"\n {self.endpoint_path}"
return context
def __str__(self) -> str:
message = f"{len(self._errors)} validation error{'s' if len(self._errors) != 1 else ''}:\n"
for err in self._errors:
message += f" {err}\n"
message += self._format_endpoint_context()
return message.rstrip()
class RequestValidationError(ValidationException):
def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:
super().__init__(errors)
def __init__(
self,
errors: Sequence[Any],
*,
body: Any = None,
endpoint_ctx: Optional[EndpointContext] = None,
) -> None:
super().__init__(errors, endpoint_ctx=endpoint_ctx)
self.body = body
class WebSocketRequestValidationError(ValidationException):
pass
def __init__(
self,
errors: Sequence[Any],
*,
endpoint_ctx: Optional[EndpointContext] = None,
) -> None:
super().__init__(errors, endpoint_ctx=endpoint_ctx)
class ResponseValidationError(ValidationException):
def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:
super().__init__(errors)
def __init__(
self,
errors: Sequence[Any],
*,
body: Any = None,
endpoint_ctx: Optional[EndpointContext] = None,
) -> None:
super().__init__(errors, endpoint_ctx=endpoint_ctx)
self.body = body
def __str__(self) -> str:
message = f"{len(self._errors)} validation errors:\n"
for err in self._errors:
message += f" {err}\n"
return message

View File

@@ -79,16 +79,25 @@ def get_openapi_security_definitions(
flat_dependant: Dependant,
) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
security_definitions = {}
operation_security = []
for security_requirement in flat_dependant.security_requirements:
# Use a dict to merge scopes for same security scheme
operation_security_dict: Dict[str, List[str]] = {}
for security_dependency in flat_dependant._security_dependencies:
security_definition = jsonable_encoder(
security_requirement.security_scheme.model,
security_dependency._security_scheme.model,
by_alias=True,
exclude_none=True,
)
security_name = security_requirement.security_scheme.scheme_name
security_name = security_dependency._security_scheme.scheme_name
security_definitions[security_name] = security_definition
operation_security.append({security_name: security_requirement.scopes})
# Merge scopes for the same security scheme
if security_name not in operation_security_dict:
operation_security_dict[security_name] = []
for scope in security_dependency.oauth_scopes or []:
if scope not in operation_security_dict[security_name]:
operation_security_dict[security_name].append(scope)
operation_security = [
{name: scopes} for name, scopes in operation_security_dict.items()
]
return security_definitions, operation_security

View File

@@ -46,6 +46,7 @@ from fastapi.dependencies.utils import (
)
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import (
EndpointContext,
FastAPIError,
RequestValidationError,
ResponseValidationError,
@@ -212,6 +213,33 @@ def _merge_lifespan_context(
return merged_lifespan # type: ignore[return-value]
# Cache for endpoint context to avoid re-extracting on every request
_endpoint_context_cache: Dict[int, 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]
try:
ctx: EndpointContext = {}
if (source_file := inspect.getsourcefile(func)) is not None:
ctx["file"] = source_file
if (line_number := inspect.getsourcelines(func)[1]) is not None:
ctx["line"] = line_number
if (func_name := getattr(func, "__name__", None)) is not None:
ctx["function"] = func_name
except Exception:
ctx = EndpointContext()
_endpoint_context_cache[func_id] = ctx
return ctx
async def serialize_response(
*,
field: Optional[ModelField] = None,
@@ -223,6 +251,7 @@ async def serialize_response(
exclude_defaults: bool = False,
exclude_none: bool = False,
is_coroutine: bool = True,
endpoint_ctx: Optional[EndpointContext] = None,
) -> Any:
if field:
errors = []
@@ -245,8 +274,11 @@ async def serialize_response(
elif errors_:
errors.append(errors_)
if errors:
ctx = endpoint_ctx or EndpointContext()
raise ResponseValidationError(
errors=_normalize_errors(errors), body=response_content
errors=_normalize_errors(errors),
body=response_content,
endpoint_ctx=ctx,
)
if hasattr(field, "serialize"):
@@ -318,6 +350,18 @@ def get_request_handler(
"fastapi_middleware_astack not found in request scope"
)
# Extract endpoint context for error messages
endpoint_ctx = (
_extract_endpoint_context(dependant.call)
if dependant.call
else EndpointContext()
)
if dependant.path:
# For mounted sub-apps, include the mount path prefix
mount_path = request.scope.get("root_path", "").rstrip("/")
endpoint_ctx["path"] = f"{request.method} {mount_path}{dependant.path}"
# Read body and auto-close files
try:
body: Any = None
@@ -355,6 +399,7 @@ def get_request_handler(
}
],
body=e.doc,
endpoint_ctx=endpoint_ctx,
)
raise validation_error from e
except HTTPException:
@@ -414,6 +459,7 @@ def get_request_handler(
exclude_defaults=response_model_exclude_defaults,
exclude_none=response_model_exclude_none,
is_coroutine=is_coroutine,
endpoint_ctx=endpoint_ctx,
)
response = actual_response_class(content, **response_args)
if not is_body_allowed_for_status_code(response.status_code):
@@ -421,7 +467,7 @@ def get_request_handler(
response.headers.raw.extend(solved_result.response.headers.raw)
if errors:
validation_error = RequestValidationError(
_normalize_errors(errors), body=body
_normalize_errors(errors), body=body, endpoint_ctx=endpoint_ctx
)
raise validation_error
@@ -438,6 +484,15 @@ def get_websocket_app(
embed_body_fields: bool = False,
) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]:
async def app(websocket: WebSocket) -> None:
endpoint_ctx = (
_extract_endpoint_context(dependant.call)
if dependant.call
else EndpointContext()
)
if dependant.path:
# For mounted sub-apps, include the mount path prefix
mount_path = websocket.scope.get("root_path", "").rstrip("/")
endpoint_ctx["path"] = f"WS {mount_path}{dependant.path}"
async_exit_stack = websocket.scope.get("fastapi_inner_astack")
assert isinstance(async_exit_stack, AsyncExitStack), (
"fastapi_inner_astack not found in request scope"
@@ -451,7 +506,8 @@ def get_websocket_app(
)
if solved_result.errors:
raise WebSocketRequestValidationError(
_normalize_errors(solved_result.errors)
_normalize_errors(solved_result.errors),
endpoint_ctx=endpoint_ctx,
)
assert dependant.call is not None, "dependant.call must be a function"
await dependant.call(**solved_result.values)

View File

@@ -132,7 +132,7 @@ def on_pre_page(page: Page, *, config: MkDocsConfig, files: Files) -> Page:
def on_page_markdown(
markdown: str, *, page: Page, config: MkDocsConfig, files: Files
) -> str:
# Set matadata["social"]["cards_layout_options"]["title"] to clean title (without
# Set metadata["social"]["cards_layout_options"]["title"] to clean title (without
# permalink)
title = page.title
clean_title = title.split("{ #")[0]

View File

@@ -48,6 +48,34 @@ async_callable_gen_dependency = AsyncCallableGenDependency()
methods_dependency = MethodsDependency()
@app.get("/callable-dependency-class")
async def get_callable_dependency_class(
value: str, instance: CallableDependency = Depends()
):
return instance(value)
@app.get("/callable-gen-dependency-class")
async def get_callable_gen_dependency_class(
value: str, instance: CallableGenDependency = Depends()
):
return next(instance(value))
@app.get("/async-callable-dependency-class")
async def get_async_callable_dependency_class(
value: str, instance: AsyncCallableDependency = Depends()
):
return await instance(value)
@app.get("/async-callable-gen-dependency-class")
async def get_async_callable_gen_dependency_class(
value: str, instance: AsyncCallableGenDependency = Depends()
):
return await instance(value).__anext__()
@app.get("/callable-dependency")
async def get_callable_dependency(value: str = Depends(callable_dependency)):
return value
@@ -114,6 +142,10 @@ client = TestClient(app)
("/synchronous-method-gen-dependency", "synchronous-method-gen-dependency"),
("/asynchronous-method-dependency", "asynchronous-method-dependency"),
("/asynchronous-method-gen-dependency", "asynchronous-method-gen-dependency"),
("/callable-dependency-class", "callable-dependency-class"),
("/callable-gen-dependency-class", "callable-gen-dependency-class"),
("/async-callable-dependency-class", "async-callable-dependency-class"),
("/async-callable-gen-dependency-class", "async-callable-gen-dependency-class"),
],
)
def test_class_dependency(route, value):

View File

@@ -1,10 +1,18 @@
import inspect
import sys
from functools import wraps
from typing import AsyncGenerator, Generator
import pytest
from fastapi import Depends, FastAPI
from fastapi.concurrency import iterate_in_threadpool, run_in_threadpool
from fastapi.testclient import TestClient
if sys.version_info >= (3, 13): # pragma: no cover
from inspect import iscoroutinefunction
else: # pragma: no cover
from asyncio import iscoroutinefunction
def noop_wrap(func):
@wraps(func)
@@ -14,8 +22,163 @@ def noop_wrap(func):
return wrapper
def noop_wrap_async(func):
if inspect.isgeneratorfunction(func):
@wraps(func)
async def gen_wrapper(*args, **kwargs):
async for item in iterate_in_threadpool(func(*args, **kwargs)):
yield item
return gen_wrapper
elif inspect.isasyncgenfunction(func):
@wraps(func)
async def async_gen_wrapper(*args, **kwargs):
async for item in func(*args, **kwargs):
yield item
return async_gen_wrapper
@wraps(func)
async def wrapper(*args, **kwargs):
if inspect.isroutine(func) and iscoroutinefunction(func):
return await func(*args, **kwargs)
if inspect.isclass(func):
return await run_in_threadpool(func, *args, **kwargs)
dunder_call = getattr(func, "__call__", None) # noqa: B004
if iscoroutinefunction(dunder_call):
return await dunder_call(*args, **kwargs)
return await run_in_threadpool(func, *args, **kwargs)
return wrapper
class ClassInstanceDep:
def __call__(self):
return True
class_instance_dep = ClassInstanceDep()
wrapped_class_instance_dep = noop_wrap(class_instance_dep)
wrapped_class_instance_dep_async_wrapper = noop_wrap_async(class_instance_dep)
class ClassInstanceGenDep:
def __call__(self):
yield True
class_instance_gen_dep = ClassInstanceGenDep()
wrapped_class_instance_gen_dep = noop_wrap(class_instance_gen_dep)
class ClassInstanceWrappedDep:
@noop_wrap
def __call__(self):
return True
class_instance_wrapped_dep = ClassInstanceWrappedDep()
class ClassInstanceWrappedAsyncDep:
@noop_wrap_async
def __call__(self):
return True
class_instance_wrapped_async_dep = ClassInstanceWrappedAsyncDep()
class ClassInstanceWrappedGenDep:
@noop_wrap
def __call__(self):
yield True
class_instance_wrapped_gen_dep = ClassInstanceWrappedGenDep()
class ClassInstanceWrappedAsyncGenDep:
@noop_wrap_async
def __call__(self):
yield True
class_instance_wrapped_async_gen_dep = ClassInstanceWrappedAsyncGenDep()
class ClassDep:
def __init__(self):
self.value = True
wrapped_class_dep = noop_wrap(ClassDep)
wrapped_class_dep_async_wrapper = noop_wrap_async(ClassDep)
class ClassInstanceAsyncDep:
async def __call__(self):
return True
class_instance_async_dep = ClassInstanceAsyncDep()
wrapped_class_instance_async_dep = noop_wrap(class_instance_async_dep)
wrapped_class_instance_async_dep_async_wrapper = noop_wrap_async(
class_instance_async_dep
)
class ClassInstanceAsyncGenDep:
async def __call__(self):
yield True
class_instance_async_gen_dep = ClassInstanceAsyncGenDep()
wrapped_class_instance_async_gen_dep = noop_wrap(class_instance_async_gen_dep)
class ClassInstanceAsyncWrappedDep:
@noop_wrap
async def __call__(self):
return True
class_instance_async_wrapped_dep = ClassInstanceAsyncWrappedDep()
class ClassInstanceAsyncWrappedAsyncDep:
@noop_wrap_async
async def __call__(self):
return True
class_instance_async_wrapped_async_dep = ClassInstanceAsyncWrappedAsyncDep()
class ClassInstanceAsyncWrappedGenDep:
@noop_wrap
async def __call__(self):
yield True
class_instance_async_wrapped_gen_dep = ClassInstanceAsyncWrappedGenDep()
class ClassInstanceAsyncWrappedGenAsyncDep:
@noop_wrap_async
async def __call__(self):
yield True
class_instance_async_wrapped_gen_async_dep = ClassInstanceAsyncWrappedGenAsyncDep()
app = FastAPI()
# Sync wrapper
@noop_wrap
def wrapped_dependency() -> bool:
@@ -59,16 +222,225 @@ async def get_async_wrapped_gen_dependency(
return value
@app.get("/wrapped-class-instance-dependency/")
async def get_wrapped_class_instance_dependency(
value: bool = Depends(wrapped_class_instance_dep),
):
return value
@app.get("/wrapped-class-instance-async-dependency/")
async def get_wrapped_class_instance_async_dependency(
value: bool = Depends(wrapped_class_instance_async_dep),
):
return value
@app.get("/wrapped-class-instance-gen-dependency/")
async def get_wrapped_class_instance_gen_dependency(
value: bool = Depends(wrapped_class_instance_gen_dep),
):
return value
@app.get("/wrapped-class-instance-async-gen-dependency/")
async def get_wrapped_class_instance_async_gen_dependency(
value: bool = Depends(wrapped_class_instance_async_gen_dep),
):
return value
@app.get("/class-instance-wrapped-dependency/")
async def get_class_instance_wrapped_dependency(
value: bool = Depends(class_instance_wrapped_dep),
):
return value
@app.get("/class-instance-wrapped-async-dependency/")
async def get_class_instance_wrapped_async_dependency(
value: bool = Depends(class_instance_wrapped_async_dep),
):
return value
@app.get("/class-instance-async-wrapped-dependency/")
async def get_class_instance_async_wrapped_dependency(
value: bool = Depends(class_instance_async_wrapped_dep),
):
return value
@app.get("/class-instance-async-wrapped-async-dependency/")
async def get_class_instance_async_wrapped_async_dependency(
value: bool = Depends(class_instance_async_wrapped_async_dep),
):
return value
@app.get("/class-instance-wrapped-gen-dependency/")
async def get_class_instance_wrapped_gen_dependency(
value: bool = Depends(class_instance_wrapped_gen_dep),
):
return value
@app.get("/class-instance-wrapped-async-gen-dependency/")
async def get_class_instance_wrapped_async_gen_dependency(
value: bool = Depends(class_instance_wrapped_async_gen_dep),
):
return value
@app.get("/class-instance-async-wrapped-gen-dependency/")
async def get_class_instance_async_wrapped_gen_dependency(
value: bool = Depends(class_instance_async_wrapped_gen_dep),
):
return value
@app.get("/class-instance-async-wrapped-gen-async-dependency/")
async def get_class_instance_async_wrapped_gen_async_dependency(
value: bool = Depends(class_instance_async_wrapped_gen_async_dep),
):
return value
@app.get("/wrapped-class-dependency/")
async def get_wrapped_class_dependency(value: ClassDep = Depends(wrapped_class_dep)):
return value.value
@app.get("/wrapped-endpoint/")
@noop_wrap
def get_wrapped_endpoint():
return True
@app.get("/async-wrapped-endpoint/")
@noop_wrap
async def get_async_wrapped_endpoint():
return True
# Async wrapper
@noop_wrap_async
def wrapped_dependency_async_wrapper() -> bool:
return True
@noop_wrap_async
def wrapped_gen_dependency_async_wrapper() -> Generator[bool, None, None]:
yield True
@noop_wrap_async
async def async_wrapped_dependency_async_wrapper() -> bool:
return True
@noop_wrap_async
async def async_wrapped_gen_dependency_async_wrapper() -> AsyncGenerator[bool, None]:
yield True
@app.get("/wrapped-dependency-async-wrapper/")
async def get_wrapped_dependency_async_wrapper(
value: bool = Depends(wrapped_dependency_async_wrapper),
):
return value
@app.get("/wrapped-gen-dependency-async-wrapper/")
async def get_wrapped_gen_dependency_async_wrapper(
value: bool = Depends(wrapped_gen_dependency_async_wrapper),
):
return value
@app.get("/async-wrapped-dependency-async-wrapper/")
async def get_async_wrapped_dependency_async_wrapper(
value: bool = Depends(async_wrapped_dependency_async_wrapper),
):
return value
@app.get("/async-wrapped-gen-dependency-async-wrapper/")
async def get_async_wrapped_gen_dependency_async_wrapper(
value: bool = Depends(async_wrapped_gen_dependency_async_wrapper),
):
return value
@app.get("/wrapped-class-instance-dependency-async-wrapper/")
async def get_wrapped_class_instance_dependency_async_wrapper(
value: bool = Depends(wrapped_class_instance_dep_async_wrapper),
):
return value
@app.get("/wrapped-class-instance-async-dependency-async-wrapper/")
async def get_wrapped_class_instance_async_dependency_async_wrapper(
value: bool = Depends(wrapped_class_instance_async_dep_async_wrapper),
):
return value
@app.get("/wrapped-class-dependency-async-wrapper/")
async def get_wrapped_class_dependency_async_wrapper(
value: ClassDep = Depends(wrapped_class_dep_async_wrapper),
):
return value.value
@app.get("/wrapped-endpoint-async-wrapper/")
@noop_wrap_async
def get_wrapped_endpoint_async_wrapper():
return True
@app.get("/async-wrapped-endpoint-async-wrapper/")
@noop_wrap_async
async def get_async_wrapped_endpoint_async_wrapper():
return True
client = TestClient(app)
@pytest.mark.parametrize(
"route",
[
"/wrapped-dependency",
"/wrapped-gen-dependency",
"/async-wrapped-dependency",
"/async-wrapped-gen-dependency",
"/wrapped-dependency/",
"/wrapped-gen-dependency/",
"/async-wrapped-dependency/",
"/async-wrapped-gen-dependency/",
"/wrapped-class-instance-dependency/",
"/wrapped-class-instance-async-dependency/",
"/wrapped-class-instance-gen-dependency/",
"/wrapped-class-instance-async-gen-dependency/",
"/class-instance-wrapped-dependency/",
"/class-instance-wrapped-async-dependency/",
"/class-instance-async-wrapped-dependency/",
"/class-instance-async-wrapped-async-dependency/",
"/class-instance-wrapped-gen-dependency/",
"/class-instance-wrapped-async-gen-dependency/",
"/class-instance-async-wrapped-gen-dependency/",
"/class-instance-async-wrapped-gen-async-dependency/",
"/wrapped-class-dependency/",
"/wrapped-endpoint/",
"/async-wrapped-endpoint/",
"/wrapped-dependency-async-wrapper/",
"/wrapped-gen-dependency-async-wrapper/",
"/async-wrapped-dependency-async-wrapper/",
"/async-wrapped-gen-dependency-async-wrapper/",
"/wrapped-class-instance-dependency-async-wrapper/",
"/wrapped-class-instance-async-dependency-async-wrapper/",
"/wrapped-class-dependency-async-wrapper/",
"/wrapped-endpoint-async-wrapper/",
"/async-wrapped-endpoint-async-wrapper/",
],
)
def test_class_dependency(route):

View File

@@ -24,6 +24,18 @@ class Item(BaseModel):
model_config = {"json_schema_serialization_defaults_required": True}
if PYDANTIC_V2:
from pydantic import computed_field
class WithComputedField(BaseModel):
name: str
@computed_field
@property
def computed_field(self) -> str:
return f"computed {self.name}"
def get_app_client(separate_input_output_schemas: bool = True) -> TestClient:
app = FastAPI(separate_input_output_schemas=separate_input_output_schemas)
@@ -46,6 +58,14 @@ def get_app_client(separate_input_output_schemas: bool = True) -> TestClient:
Item(name="Plumbus"),
]
if PYDANTIC_V2:
@app.post("/with-computed-field/")
def create_with_computed_field(
with_computed_field: WithComputedField,
) -> WithComputedField:
return with_computed_field
client = TestClient(app)
return client
@@ -131,6 +151,23 @@ def test_read_items():
)
@needs_pydanticv2
def test_with_computed_field():
client = get_app_client()
client_no = get_app_client(separate_input_output_schemas=False)
response = client.post("/with-computed-field/", json={"name": "example"})
response2 = client_no.post("/with-computed-field/", json={"name": "example"})
assert response.status_code == response2.status_code == 200, response.text
assert (
response.json()
== response2.json()
== {
"name": "example",
"computed_field": "computed example",
}
)
@needs_pydanticv2
def test_openapi_schema():
client = get_app_client()
@@ -245,6 +282,44 @@ def test_openapi_schema():
},
}
},
"/with-computed-field/": {
"post": {
"summary": "Create With Computed Field",
"operationId": "create_with_computed_field_with_computed_field__post",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/WithComputedField-Input"
}
}
},
"required": True,
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/WithComputedField-Output"
}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
},
},
"components": {
"schemas": {
@@ -333,6 +408,25 @@ def test_openapi_schema():
"required": ["subname", "sub_description", "tags"],
"title": "SubItem",
},
"WithComputedField-Input": {
"properties": {"name": {"type": "string", "title": "Name"}},
"type": "object",
"required": ["name"],
"title": "WithComputedField",
},
"WithComputedField-Output": {
"properties": {
"name": {"type": "string", "title": "Name"},
"computed_field": {
"type": "string",
"title": "Computed Field",
"readOnly": True,
},
},
"type": "object",
"required": ["name", "computed_field"],
"title": "WithComputedField",
},
"ValidationError": {
"properties": {
"loc": {
@@ -458,6 +552,44 @@ def test_openapi_schema_no_separate():
},
}
},
"/with-computed-field/": {
"post": {
"summary": "Create With Computed Field",
"operationId": "create_with_computed_field_with_computed_field__post",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/WithComputedField-Input"
}
}
},
"required": True,
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/WithComputedField-Output"
}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
},
},
"components": {
"schemas": {
@@ -508,6 +640,25 @@ def test_openapi_schema_no_separate():
"required": ["subname"],
"title": "SubItem",
},
"WithComputedField-Input": {
"properties": {"name": {"type": "string", "title": "Name"}},
"type": "object",
"required": ["name"],
"title": "WithComputedField",
},
"WithComputedField-Output": {
"properties": {
"name": {"type": "string", "title": "Name"},
"computed_field": {
"type": "string",
"title": "Computed Field",
"readOnly": True,
},
},
"type": "object",
"required": ["name", "computed_field"],
"title": "WithComputedField",
},
"ValidationError": {
"properties": {
"loc": {

View File

@@ -0,0 +1,198 @@
# Ref: https://github.com/fastapi/fastapi/issues/14454
from typing import Optional
from fastapi import APIRouter, Depends, FastAPI, Security
from fastapi.security import OAuth2AuthorizationCodeBearer
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from typing_extensions import Annotated
oauth2_scheme = OAuth2AuthorizationCodeBearer(
authorizationUrl="authorize",
tokenUrl="token",
auto_error=True,
scopes={"read": "Read access", "write": "Write access"},
)
async def get_token(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
return token
app = FastAPI(dependencies=[Depends(get_token)])
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get(
"/with-oauth2-scheme",
dependencies=[Security(oauth2_scheme, scopes=["read", "write"])],
)
async def read_with_oauth2_scheme():
return {"message": "Admin Access"}
@app.get(
"/with-get-token", dependencies=[Security(get_token, scopes=["read", "write"])]
)
async def read_with_get_token():
return {"message": "Admin Access"}
router = APIRouter(dependencies=[Security(oauth2_scheme, scopes=["read"])])
@router.get("/items/")
async def read_items(token: Optional[str] = Depends(oauth2_scheme)):
return {"token": token}
@router.post("/items/")
async def create_item(
token: Optional[str] = Security(oauth2_scheme, scopes=["read", "write"]),
):
return {"token": token}
app.include_router(router)
client = TestClient(app)
def test_root():
response = client.get("/", headers={"Authorization": "Bearer testtoken"})
assert response.status_code == 200, response.text
assert response.json() == {"message": "Hello World"}
def test_read_with_oauth2_scheme():
response = client.get(
"/with-oauth2-scheme", headers={"Authorization": "Bearer testtoken"}
)
assert response.status_code == 200, response.text
assert response.json() == {"message": "Admin Access"}
def test_read_with_get_token():
response = client.get(
"/with-get-token", headers={"Authorization": "Bearer testtoken"}
)
assert response.status_code == 200, response.text
assert response.json() == {"message": "Admin Access"}
def test_read_token():
response = client.get("/items/", headers={"Authorization": "Bearer testtoken"})
assert response.status_code == 200, response.text
assert response.json() == {"token": "testtoken"}
def test_create_token():
response = client.post("/items/", headers={"Authorization": "Bearer testtoken"})
assert response.status_code == 200, response.text
assert response.json() == {"token": "testtoken"}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/": {
"get": {
"summary": "Root",
"operationId": "root__get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [{"OAuth2AuthorizationCodeBearer": []}],
}
},
"/with-oauth2-scheme": {
"get": {
"summary": "Read With Oauth2 Scheme",
"operationId": "read_with_oauth2_scheme_with_oauth2_scheme_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [
{"OAuth2AuthorizationCodeBearer": ["read", "write"]}
],
}
},
"/with-get-token": {
"get": {
"summary": "Read With Get Token",
"operationId": "read_with_get_token_with_get_token_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [
{"OAuth2AuthorizationCodeBearer": ["read", "write"]}
],
}
},
"/items/": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [
{"OAuth2AuthorizationCodeBearer": ["read"]},
],
},
"post": {
"summary": "Create Item",
"operationId": "create_item_items__post",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [
{"OAuth2AuthorizationCodeBearer": ["read", "write"]},
],
},
},
},
"components": {
"securitySchemes": {
"OAuth2AuthorizationCodeBearer": {
"type": "oauth2",
"flows": {
"authorizationCode": {
"scopes": {
"read": "Read access",
"write": "Write access",
},
"authorizationUrl": "authorize",
"tokenUrl": "token",
}
},
}
}
},
}
)

View File

@@ -0,0 +1,79 @@
# Ref: https://github.com/fastapi/fastapi/issues/14454
from fastapi import Depends, FastAPI, Security
from fastapi.security import OAuth2AuthorizationCodeBearer
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from typing_extensions import Annotated
oauth2_scheme = OAuth2AuthorizationCodeBearer(
authorizationUrl="api/oauth/authorize",
tokenUrl="/api/oauth/token",
scopes={"read": "Read access", "write": "Write access"},
)
async def get_token(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
return token
app = FastAPI(dependencies=[Depends(get_token)])
@app.get("/admin", dependencies=[Security(get_token, scopes=["read", "write"])])
async def read_admin():
return {"message": "Admin Access"}
client = TestClient(app)
def test_read_admin():
response = client.get("/admin", headers={"Authorization": "Bearer faketoken"})
assert response.status_code == 200, response.text
assert response.json() == {"message": "Admin Access"}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/admin": {
"get": {
"summary": "Read Admin",
"operationId": "read_admin_admin_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [
{"OAuth2AuthorizationCodeBearer": ["read", "write"]}
],
}
}
},
"components": {
"securitySchemes": {
"OAuth2AuthorizationCodeBearer": {
"type": "oauth2",
"flows": {
"authorizationCode": {
"scopes": {
"read": "Read access",
"write": "Write access",
},
"authorizationUrl": "api/oauth/authorize",
"tokenUrl": "/api/oauth/token",
}
},
}
}
},
}
)

View File

@@ -0,0 +1,26 @@
from __future__ import annotations
from fastapi import Depends, FastAPI, Request
from fastapi.testclient import TestClient
from typing_extensions import Annotated
from .utils import needs_py310
class Dep:
def __call__(self, request: Request):
return "test"
@needs_py310
def test_stringified_annotations():
app = FastAPI()
client = TestClient(app)
@app.get("/test/")
def call(test: Annotated[str, Depends(Dep())]):
return {"test": test}
response = client.get("/test")
assert response.status_code == 200

View File

@@ -0,0 +1,168 @@
from fastapi import FastAPI, Request, WebSocket
from fastapi.exceptions import (
RequestValidationError,
ResponseValidationError,
WebSocketRequestValidationError,
)
from fastapi.testclient import TestClient
from pydantic import BaseModel
class Item(BaseModel):
id: int
name: str
class ExceptionCapture:
def __init__(self):
self.exception = None
def capture(self, exc):
self.exception = exc
return exc
app = FastAPI()
sub_app = FastAPI()
captured_exception = ExceptionCapture()
app.mount(path="/sub", app=sub_app)
@app.exception_handler(RequestValidationError)
@sub_app.exception_handler(RequestValidationError)
async def request_validation_handler(request: Request, exc: RequestValidationError):
captured_exception.capture(exc)
raise exc
@app.exception_handler(ResponseValidationError)
@sub_app.exception_handler(ResponseValidationError)
async def response_validation_handler(_: Request, exc: ResponseValidationError):
captured_exception.capture(exc)
raise exc
@app.exception_handler(WebSocketRequestValidationError)
@sub_app.exception_handler(WebSocketRequestValidationError)
async def websocket_validation_handler(
websocket: WebSocket, exc: WebSocketRequestValidationError
):
captured_exception.capture(exc)
raise exc
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"user_id": user_id} # pragma: no cover
@app.get("/items/", response_model=Item)
def get_item():
return {"name": "Widget"}
@sub_app.get("/items/", response_model=Item)
def get_sub_item():
return {"name": "Widget"} # pragma: no cover
@app.websocket("/ws/{item_id}")
async def websocket_endpoint(websocket: WebSocket, item_id: int):
await websocket.accept() # pragma: no cover
await websocket.send_text(f"Item: {item_id}") # pragma: no cover
await websocket.close() # pragma: no cover
@sub_app.websocket("/ws/{item_id}")
async def subapp_websocket_endpoint(websocket: WebSocket, item_id: int):
await websocket.accept() # pragma: no cover
await websocket.send_text(f"Item: {item_id}") # pragma: no cover
await websocket.close() # pragma: no cover
client = TestClient(app)
def test_request_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
client.get("/users/invalid")
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "get_user" in error_str
assert "/users/" in error_str
def test_response_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
client.get("/items/")
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "get_item" in error_str
assert "/items/" in error_str
def test_websocket_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
with client.websocket_connect("/ws/invalid"):
pass # pragma: no cover
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "websocket_endpoint" in error_str
assert "/ws/" in error_str
def test_subapp_request_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
client.get("/sub/items/")
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "get_sub_item" in error_str
assert "/sub/items/" in error_str
def test_subapp_websocket_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
with client.websocket_connect("/sub/ws/invalid"):
pass # pragma: no cover
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "subapp_websocket_endpoint" in error_str
assert "/sub/ws/" in error_str
def test_validation_error_with_only_path():
errors = [{"type": "missing", "loc": ("body", "name"), "msg": "Field required"}]
exc = RequestValidationError(errors, endpoint_ctx={"path": "GET /api/test"})
error_str = str(exc)
assert "Endpoint: GET /api/test" in error_str
assert 'File "' not in error_str
def test_validation_error_with_no_context():
errors = [{"type": "missing", "loc": ("body", "name"), "msg": "Field required"}]
exc = RequestValidationError(errors, endpoint_ctx={})
error_str = str(exc)
assert "1 validation error:" in error_str
assert "Endpoint" not in error_str
assert 'File "' not in error_str