Browse Source

Merge 5158c48e4b into 1d434dec47

pull/10861/merge
whysage 3 days ago
committed by GitHub
parent
commit
7408af2799
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. 76
      fastapi/applications.py
  2. 108
      fastapi/routing.py
  3. 362
      tests/test_auto_options_route.py

76
fastapi/applications.py

@ -1084,6 +1084,7 @@ class FastAPI(Starlette):
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
generate_unique_id
),
add_auto_options_route: bool = False,
) -> None:
self.router.add_api_route(
path,
@ -1110,6 +1111,7 @@ class FastAPI(Starlette):
name=name,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def api_route(
@ -1140,6 +1142,7 @@ class FastAPI(Starlette):
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
generate_unique_id
),
add_auto_options_route: bool = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
def decorator(func: DecoratedCallable) -> DecoratedCallable:
self.router.add_api_route(
@ -1167,6 +1170,7 @@ class FastAPI(Starlette):
name=name,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
return func
@ -1788,6 +1792,14 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP GET operation.
@ -1828,6 +1840,7 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def put(
@ -2161,6 +2174,14 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP PUT operation.
@ -2206,6 +2227,7 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def post(
@ -2539,6 +2561,14 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP POST operation.
@ -2584,6 +2614,7 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def delete(
@ -2917,6 +2948,14 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP DELETE operation.
@ -2957,6 +2996,7 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def options(
@ -3290,6 +3330,14 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP OPTIONS operation.
@ -3330,6 +3378,7 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def head(
@ -3663,6 +3712,14 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP HEAD operation.
@ -3703,6 +3760,7 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def patch(
@ -4036,6 +4094,14 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP PATCH operation.
@ -4081,6 +4147,7 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def trace(
@ -4414,6 +4481,14 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP TRACE operation.
@ -4454,6 +4529,7 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def websocket_route(

108
fastapi/routing.py

@ -459,6 +459,7 @@ class APIRoute(routing.Route):
generate_unique_id_function: Union[
Callable[["APIRoute"], str], DefaultPlaceholder
] = Default(generate_unique_id),
is_auto_options: bool = False,
) -> None:
self.path = path
self.endpoint = endpoint
@ -567,6 +568,7 @@ class APIRoute(routing.Route):
embed_body_fields=self._embed_body_fields,
)
self.app = request_response(self.get_route_handler())
self.is_auto_options = is_auto_options
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
return get_request_handler(
@ -911,6 +913,7 @@ class APIRouter(routing.Router):
generate_unique_id_function: Union[
Callable[[APIRoute], str], DefaultPlaceholder
] = Default(generate_unique_id),
add_auto_options_route: Optional[bool] = False,
) -> None:
route_class = route_class_override or self.route_class
responses = responses or {}
@ -959,6 +962,37 @@ class APIRouter(routing.Router):
generate_unique_id_function=current_generate_unique_id,
)
self.routes.append(route)
if add_auto_options_route:
self._update_auto_options_routes(route, path)
def _update_auto_options_routes(self, new_route: APIRoute, path: str) -> None:
auto_options_index: Optional[int] = None
allowed_methods: Set[str] = set()
for index, route in enumerate(self.routes):
if isinstance(route, APIRoute):
if route.path == new_route.path:
if hasattr(route, "is_auto_options") and route.is_auto_options:
auto_options_index = index
else:
allowed_methods.update(route.methods)
if auto_options_index is not None:
self.routes.pop(auto_options_index)
if "OPTIONS" not in new_route.methods:
async def options_route() -> Response:
return Response(headers={"Allow": ", ".join(allowed_methods)})
self.routes.append(
APIRoute(
self.prefix + path,
endpoint=options_route,
methods=["OPTIONS"],
include_in_schema=True,
is_auto_options=True,
)
)
def api_route(
self,
@ -989,6 +1023,7 @@ class APIRouter(routing.Router):
generate_unique_id_function: Callable[[APIRoute], str] = Default(
generate_unique_id
),
add_auto_options_route: Optional[bool] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
def decorator(func: DecoratedCallable) -> DecoratedCallable:
self.add_api_route(
@ -1017,6 +1052,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
return func
@ -1694,6 +1730,14 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP GET operation.
@ -1738,6 +1782,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def put(
@ -2071,6 +2116,14 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP PUT operation.
@ -2120,6 +2173,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def post(
@ -2453,6 +2507,14 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP POST operation.
@ -2502,6 +2564,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def delete(
@ -2835,6 +2898,14 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP DELETE operation.
@ -2879,6 +2950,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def options(
@ -3212,6 +3284,14 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP OPTIONS operation.
@ -3256,6 +3336,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def head(
@ -3589,6 +3670,14 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP HEAD operation.
@ -3638,6 +3727,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def patch(
@ -3971,6 +4061,14 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP PATCH operation.
@ -4020,6 +4118,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
def trace(
@ -4353,6 +4452,14 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
add_auto_options_route: Annotated[
Optional[bool],
Doc(
"""
Automatically create a route to handle the OPTIONS HTTP verb.
"""
),
] = False,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP TRACE operation.
@ -4402,6 +4509,7 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
add_auto_options_route=add_auto_options_route,
)
@deprecated(

362
tests/test_auto_options_route.py

@ -0,0 +1,362 @@
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
@app.get("/home", add_auto_options_route=True)
def get_home():
return {"hello": "world"}
@app.post("/items/", add_auto_options_route=True)
def create_item(item: Item):
return item
@app.delete("/items/{item_id}")
def delete_item(item_id: str, item: Item):
return {"item_id": item_id, "item": item}
@app.head("/items/{item_id}", add_auto_options_route=True)
def head_item(item_id: str):
return JSONResponse(None, headers={"x-fastapi-item-id": item_id})
@app.patch("/items/{item_id}", add_auto_options_route=True)
def patch_item(item_id: str, item: Item):
return {"item_id": item_id, "item": item}
@app.trace("/items/{item_id}", add_auto_options_route=True)
def trace_item(item_id: str):
return JSONResponse(None, media_type="message/http")
client = TestClient(app)
def test_get_api_route():
response = client.get("/home")
assert response.status_code == 200, response.text
assert response.json() == {"hello": "world"}
def test_post_api_route():
response = client.post("/items/", json={"name": "CoolItem"})
assert response.status_code == 200, response.text
assert response.json() == {"name": "CoolItem"}
def test_delete():
response = client.request("DELETE", "/items/foo", json={"name": "Foo"})
assert response.status_code == 200, response.text
assert response.json() == {"item_id": "foo", "item": {"name": "Foo"}}
def test_head():
response = client.head("/items/foo")
assert response.status_code == 200, response.text
assert response.headers["x-fastapi-item-id"] == "foo"
def test_patch():
response = client.patch("/items/foo", json={"name": "Foo"})
assert response.status_code == 200, response.text
assert response.json() == {"item_id": "foo", "item": {"name": "Foo"}}
def test_trace():
response = client.request("trace", "/items/foo")
assert response.status_code == 200, response.text
assert response.headers["content-type"] == "message/http"
def test_get_auto_options():
response = client.options("/home")
assert response.status_code == 200, response.text
assert response.headers.raw[0][0].decode("utf-8") == "allow"
assert response.headers.raw[0][1].decode("utf-8") == "GET"
def test_post_auto_options():
response = client.options("/items/")
assert response.status_code == 200, response.text
assert response.headers.raw[0][0].decode("utf-8") == "allow"
assert response.headers.raw[0][1].decode("utf-8") == "POST"
def test_head_auto_options():
response = client.head("/items/foo")
assert response.status_code == 200, response.text
assert response.headers["x-fastapi-item-id"] == "foo"
def test_other_auto_options():
response = client.options("/items/foo")
assert response.status_code == 200, response.text
assert response.headers.raw[0][0].decode("utf-8") == "allow"
assert set(response.headers.raw[0][1].decode("utf-8").split(", ")) == {
"DELETE",
"HEAD",
"PATCH",
"TRACE",
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == {
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/home": {
"get": {
"summary": "Get Home",
"operationId": "get_home_home_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
},
"options": {
"summary": "Options Route",
"operationId": "options_route_home_options",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
},
},
"/items/": {
"post": {
"summary": "Create Item",
"operationId": "create_item_items__post",
"requestBody": {
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"}
}
},
"required": True,
},
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
"options": {
"summary": "Options Route",
"operationId": "options_route_items__options",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
},
},
"/items/{item_id}": {
"delete": {
"summary": "Delete Item",
"operationId": "delete_item_items__item_id__delete",
"parameters": [
{
"name": "item_id",
"in": "path",
"required": True,
"schema": {"type": "string", "title": "Item Id"},
}
],
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"}
}
},
},
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
"head": {
"summary": "Head Item",
"operationId": "head_item_items__item_id__head",
"parameters": [
{
"name": "item_id",
"in": "path",
"required": True,
"schema": {"type": "string", "title": "Item Id"},
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
"patch": {
"summary": "Patch Item",
"operationId": "patch_item_items__item_id__patch",
"parameters": [
{
"name": "item_id",
"in": "path",
"required": True,
"schema": {"type": "string", "title": "Item Id"},
}
],
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"}
}
},
},
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
"trace": {
"summary": "Trace Item",
"operationId": "trace_item_items__item_id__trace",
"parameters": [
{
"name": "item_id",
"in": "path",
"required": True,
"schema": {"type": "string", "title": "Item Id"},
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
"options": {
"summary": "Options Route",
"operationId": "options_route_items__item_id__options",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
},
},
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {"$ref": "#/components/schemas/ValidationError"},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"Item": {
"properties": {"name": {"type": "string", "title": "Name"}},
"type": "object",
"required": ["name"],
"title": "Item",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
Loading…
Cancel
Save