Merge branch 'master' into improved_tracebacks

This commit is contained in:
Sebastián Ramírez 2025-12-06 04:18:53 -08:00 committed by GitHub
commit 1025e43876
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 562 additions and 39 deletions

View File

@ -7,6 +7,25 @@ hide:
## Latest Changes
## 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.123.10"
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

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

@ -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",
}
},
}
}
},
}
)