mirror of https://github.com/tiangolo/fastapi.git
Merge e62a4d444e into 272204c0c7
This commit is contained in:
commit
7e77d865fa
|
|
@ -109,6 +109,37 @@ You could also use `from starlette.requests import Request` and `from starlette.
|
|||
|
||||
///
|
||||
|
||||
### Handle multiple exceptions or status codes
|
||||
|
||||
You can register the same handler for multiple exceptions or multiple status codes at once. Just pass a list or tuple of them to `@app.exception_handler(...)`.
|
||||
|
||||
This is useful when you want to group related errors together and respond with the same logic.
|
||||
|
||||
For example, if you want to treat 401 Unauthorized and 403 Forbidden as access-related issues:
|
||||
|
||||
{* ../../docs_src/handling_errors/tutorial007.py hl[15:20,33,37] *}
|
||||
|
||||
Raising an `HTTPException` with either a `401` or `403` status code will result in the same response detail:
|
||||
|
||||
```JSON
|
||||
{"detail": "Access denied. Check your credentials or permissions."}
|
||||
```
|
||||
|
||||
Or you can handle multiple exception classes like this:
|
||||
|
||||
{* ../../docs_src/handling_errors/tutorial008.py hl[10:12,15:17,20:25,32,38] *}
|
||||
|
||||
Here, if your request causes either a `FileTooLargeError` or an `UnsupportedFileTypeError`, the `custom_exception_handler` will be used to handle the exception and add a `hint` field to the response:
|
||||
|
||||
```JSON
|
||||
{
|
||||
"error": "The uploaded file is too large.",
|
||||
"hint": "Need help? Contact support@example.com"
|
||||
}
|
||||
```
|
||||
|
||||
This allows for simpler, more maintainable error handling when several conditions should result in the same kind of response.
|
||||
|
||||
## Override the default exception handlers { #override-the-default-exception-handlers }
|
||||
|
||||
**FastAPI** has some default exception handlers.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,39 @@
|
|||
from typing import Union
|
||||
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
FAKE_DB = {
|
||||
0: {"name": "Admin", "role": "ADMIN"},
|
||||
1: {"name": "User 1", "role": "USER"},
|
||||
2: {"name": "User 2", "role": "USER"},
|
||||
}
|
||||
|
||||
|
||||
@app.exception_handler([401, 403])
|
||||
async def handle_auth_errors(request: Request, exc: Exception):
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code if isinstance(exc, HTTPException) else 403,
|
||||
content={"detail": "Access denied. Check your credentials or permissions."},
|
||||
)
|
||||
|
||||
|
||||
@app.get("/secrets/")
|
||||
async def get_secrets(auth_user_id: Union[int, None] = None):
|
||||
# Get authenticated user info (not a production-ready code)
|
||||
if auth_user_id is not None:
|
||||
auth_user_info = FAKE_DB.get(auth_user_id)
|
||||
else:
|
||||
auth_user_info = None
|
||||
|
||||
# Return 401 status code if user not authenticated
|
||||
if auth_user_info is None:
|
||||
raise HTTPException(status_code=401) # Not authenticated
|
||||
|
||||
# Return 403 status code if user is not authorized to get secret information
|
||||
if auth_user_info["role"] != "ADMIN":
|
||||
raise HTTPException(status_code=403) # Not authorized
|
||||
|
||||
return {"data": "Secret information"}
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
from fastapi import FastAPI, File, HTTPException, Request, UploadFile
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
MAX_FILE_SIZE_MB = 5
|
||||
ALLOWED_TYPES = {"application/pdf", "image/jpeg"}
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
class FileTooLargeError(HTTPException):
|
||||
def __init__(self):
|
||||
super().__init__(status_code=413, detail="The uploaded file is too large.")
|
||||
|
||||
|
||||
class UnsupportedFileTypeError(HTTPException):
|
||||
def __init__(self):
|
||||
super().__init__(status_code=415, detail="Unsupported file type")
|
||||
|
||||
|
||||
@app.exception_handler((FileTooLargeError, UnsupportedFileTypeError))
|
||||
async def custom_exception_handler(request: Request, exc: HTTPException):
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content={"error": exc.detail, "hint": "Need help? Contact support@example.com"},
|
||||
)
|
||||
|
||||
|
||||
@app.post("/upload/")
|
||||
async def upload_file(file: UploadFile = File(...)):
|
||||
# Validate file type
|
||||
if file.content_type not in ALLOWED_TYPES:
|
||||
raise UnsupportedFileTypeError()
|
||||
|
||||
# Validate file size (read contents to check size in memory)
|
||||
contents = await file.read()
|
||||
size_mb = len(contents) / (1024 * 1024)
|
||||
if size_mb > MAX_FILE_SIZE_MB:
|
||||
raise FileTooLargeError()
|
||||
|
||||
return {"filename": file.filename, "message": "File uploaded successfully!"}
|
||||
|
|
@ -4628,10 +4628,10 @@ class FastAPI(Starlette):
|
|||
def exception_handler(
|
||||
self,
|
||||
exc_class_or_status_code: Annotated[
|
||||
Union[int, Type[Exception]],
|
||||
Union[int, Type[Exception], Sequence[int], Sequence[Type[Exception]]],
|
||||
Doc(
|
||||
"""
|
||||
The Exception class this would handle, or a status code.
|
||||
The Exception class, a status code or a sequence of them this would handle.
|
||||
"""
|
||||
),
|
||||
],
|
||||
|
|
@ -4667,7 +4667,11 @@ class FastAPI(Starlette):
|
|||
"""
|
||||
|
||||
def decorator(func: DecoratedCallable) -> DecoratedCallable:
|
||||
self.add_exception_handler(exc_class_or_status_code, func)
|
||||
if isinstance(exc_class_or_status_code, Sequence):
|
||||
for exc_class_or_status_code_ in exc_class_or_status_code:
|
||||
self.add_exception_handler(exc_class_or_status_code_, func)
|
||||
else:
|
||||
self.add_exception_handler(exc_class_or_status_code, func)
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
|
|
|||
|
|
@ -5,6 +5,14 @@ from fastapi.testclient import TestClient
|
|||
from starlette.responses import JSONResponse
|
||||
|
||||
|
||||
class CustomException1(HTTPException):
|
||||
pass
|
||||
|
||||
|
||||
class CustomException2(HTTPException):
|
||||
pass
|
||||
|
||||
|
||||
def http_exception_handler(request, exception):
|
||||
return JSONResponse({"exception": "http-exception"})
|
||||
|
||||
|
|
@ -86,3 +94,62 @@ def test_traceback_for_dependency_with_yield():
|
|||
last_frame = exc_info.traceback[-1]
|
||||
assert str(last_frame.path) == __file__
|
||||
assert last_frame.lineno == raise_value_error.__code__.co_firstlineno
|
||||
|
||||
|
||||
def test_exception_handler_with_single_exception():
|
||||
local_app = FastAPI()
|
||||
|
||||
@local_app.exception_handler(CustomException1)
|
||||
def custom_exception_handler(request, exception):
|
||||
pass # pragma: no cover
|
||||
|
||||
assert (
|
||||
local_app.exception_handlers.get(CustomException1) == custom_exception_handler
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exceptions",
|
||||
[
|
||||
(CustomException1, CustomException2), # Tuple of exceptions
|
||||
[CustomException1, CustomException2], # List of exceptions
|
||||
],
|
||||
)
|
||||
def test_exception_handler_with_multiple_exceptions(exceptions):
|
||||
local_app = FastAPI()
|
||||
|
||||
@local_app.exception_handler(exceptions)
|
||||
def custom_exception_handler(request, exception):
|
||||
pass # pragma: no cover
|
||||
|
||||
assert local_app.exception_handlers.get(exceptions[0]) == custom_exception_handler
|
||||
|
||||
assert local_app.exception_handlers.get(exceptions[1]) == custom_exception_handler
|
||||
|
||||
|
||||
def test_exception_handler_with_single_status_code():
|
||||
local_app = FastAPI()
|
||||
|
||||
@local_app.exception_handler(409)
|
||||
def http_409_status_code_handler(request, exception):
|
||||
pass # pragma: no cover
|
||||
|
||||
assert local_app.exception_handlers.get(409) == http_409_status_code_handler
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"status_codes",
|
||||
[
|
||||
(401, 403), # Tuple of status codes
|
||||
[401, 403], # List of status codes
|
||||
],
|
||||
)
|
||||
def test_exception_handler_with_multiple_status_codes(status_codes):
|
||||
local_app = FastAPI()
|
||||
|
||||
@local_app.exception_handler(status_codes)
|
||||
def auth_errors_handler(request, exception):
|
||||
pass # pragma: no cover
|
||||
|
||||
assert local_app.exception_handlers.get(status_codes[0]) == auth_errors_handler
|
||||
assert local_app.exception_handlers.get(status_codes[1]) == auth_errors_handler
|
||||
|
|
|
|||
|
|
@ -0,0 +1,27 @@
|
|||
from fastapi.testclient import TestClient
|
||||
|
||||
from docs_src.handling_errors.tutorial007 import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_unauthenticated():
|
||||
response = client.get("/secrets")
|
||||
assert response.status_code == 401, response.text
|
||||
assert response.json() == {
|
||||
"detail": "Access denied. Check your credentials or permissions."
|
||||
}
|
||||
|
||||
|
||||
def test_unauthorized():
|
||||
response = client.get("/secrets", params={"auth_user_id": 1})
|
||||
assert response.status_code == 403, response.text
|
||||
assert response.json() == {
|
||||
"detail": "Access denied. Check your credentials or permissions."
|
||||
}
|
||||
|
||||
|
||||
def test_success():
|
||||
response = client.get("/secrets", params={"auth_user_id": 0})
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json() == {"data": "Secret information"}
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from docs_src.handling_errors import tutorial008
|
||||
from docs_src.handling_errors.tutorial008 import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_unsupported_file_type(tmp_path: Path):
|
||||
file = tmp_path / "test.txt"
|
||||
file.write_text("<file content>")
|
||||
with open(file, "+rb") as fp:
|
||||
response = client.post(
|
||||
"/upload",
|
||||
files={"file": ("test.txt", fp, "text/plain")},
|
||||
)
|
||||
assert response.status_code == 415, response.text
|
||||
assert response.json() == {
|
||||
"error": "Unsupported file type",
|
||||
"hint": "Need help? Contact support@example.com",
|
||||
}
|
||||
|
||||
|
||||
def test_file_too_large(tmp_path: Path):
|
||||
file = tmp_path / "test.pdf"
|
||||
file.write_text("<file content>" * 100) # ~1.37 kB
|
||||
with patch.object(
|
||||
tutorial008,
|
||||
"MAX_FILE_SIZE_MB",
|
||||
new=0.001, # MAX_FILE_SIZE_MB = 1 kB
|
||||
):
|
||||
with open(file, "+rb") as fp:
|
||||
response = client.post(
|
||||
"/upload",
|
||||
files={"file": ("test.pdf", fp, "application/pdf")},
|
||||
)
|
||||
assert response.status_code == 413, response.text
|
||||
assert response.json() == {
|
||||
"error": "The uploaded file is too large.",
|
||||
"hint": "Need help? Contact support@example.com",
|
||||
}
|
||||
|
||||
|
||||
def test_success(tmp_path: Path):
|
||||
file = tmp_path / "test.pdf"
|
||||
file.write_text("<file content>")
|
||||
with open(file, "+rb") as fp:
|
||||
response = client.post(
|
||||
"/upload",
|
||||
files={"file": ("test.pdf", fp, "application/pdf")},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json() == {
|
||||
"filename": "test.pdf",
|
||||
"message": "File uploaded successfully!",
|
||||
}
|
||||
Loading…
Reference in New Issue