Compare commits

...

16 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
14 changed files with 862 additions and 56 deletions

View File

@@ -7,6 +7,35 @@ 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

View File

@@ -1,6 +1,6 @@
"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
__version__ = "0.123.7"
__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
@@ -28,12 +28,6 @@ def _impartial(func: Callable[..., Any]) -> Callable[..., Any]:
return func
@dataclass
class SecurityRequirement:
security_scheme: SecurityBase
scopes: Optional[Sequence[str]] = None
@dataclass
class Dependant:
path_params: List[ModelField] = field(default_factory=list)
@@ -42,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
@@ -83,11 +76,32 @@ 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 _is_security_scheme(self) -> bool:
if self.call is None:
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 self.call is None:
@@ -96,6 +110,8 @@ class Dependant:
_impartial(self.call)
) or inspect.isgeneratorfunction(_unwrapped_call(self.call)):
return True
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
@@ -120,6 +136,8 @@ class Dependant:
_impartial(self.call)
) or inspect.isasyncgenfunction(_unwrapped_call(self.call)):
return True
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
@@ -148,6 +166,8 @@ class Dependant:
_unwrapped_call(self.call)
):
return True
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
@@ -162,7 +182,6 @@ class Dependant:
_impartial(dunder_unwrapped_call)
) or iscoroutinefunction(_unwrapped_call(dunder_unwrapped_call)):
return True
# if inspect.isclass(self.call): False, covered by default return
return False
@cached_property

View File

@@ -55,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
@@ -142,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(),
@@ -153,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
@@ -258,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(

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

@@ -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,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