Browse Source
Allow scoping exception handlers to specific routers instead of only at the app level. Router-level handlers override app-level handlers for routes within that router, with precedence: route > child router > parent router > app-level middleware.pull/15067/head
Michał Połtyn
4 months ago
committed by
Michał Połtyn
Michał Połtyn
10 changed files with 827 additions and 2 deletions
@ -0,0 +1,79 @@ |
|||||
|
# Router-Level Exception Handlers { #router-level-exception-handlers } |
||||
|
|
||||
|
In the [Tutorial - Handling Errors](../tutorial/handling-errors.md){.internal-link target=_blank} you learned how to add custom exception handlers to your **FastAPI** application using `@app.exception_handler()`. |
||||
|
|
||||
|
Those handlers apply **globally** to all routes in the application. |
||||
|
|
||||
|
But sometimes you want to handle exceptions differently depending on which part of your application raised them. For example, a group of routes related to payments might need different error handling than routes for user profiles. |
||||
|
|
||||
|
You can do this by adding **exception handlers directly to an `APIRouter`**. |
||||
|
|
||||
|
## Add Exception Handlers to a Router { #add-exception-handlers-to-a-router } |
||||
|
|
||||
|
You can pass `exception_handlers` when creating an `APIRouter`, using the same format as with the `FastAPI` app — a dictionary where keys are exception classes (or status codes) and values are handler functions: |
||||
|
|
||||
|
{* ../../docs_src/handling_errors/tutorial007_py310.py hl[10:20] *} |
||||
|
|
||||
|
Now, if a `UnicornException` is raised in any route within this router, the router's handler will catch it. |
||||
|
|
||||
|
Routes outside this router are **not affected** by the router's exception handlers. |
||||
|
|
||||
|
## Router Handlers Override App Handlers { #router-handlers-override-app-handlers } |
||||
|
|
||||
|
If both the app and a router define a handler for the same exception, the **router's handler takes priority** for routes within that router. The app-level handler still applies to all other routes. |
||||
|
|
||||
|
{* ../../docs_src/handling_errors/tutorial008_py310.py hl[13:21,44:49] *} |
||||
|
|
||||
|
In this example: |
||||
|
|
||||
|
* A request to `/magic/unicorns/yolo` uses the **global** handler (the `magic_router` doesn't define its own). |
||||
|
* A request to `/special/unicorns/yolo` uses the **router-level** handler (defined on the `special_router`). |
||||
|
|
||||
|
This lets you customize error handling per section of your API while keeping a sensible default at the app level. |
||||
|
|
||||
|
## Using `add_exception_handler` { #using-add-exception-handler } |
||||
|
|
||||
|
You can also add exception handlers to a router after creation, using the `add_exception_handler()` method: |
||||
|
|
||||
|
```python |
||||
|
router = APIRouter() |
||||
|
router.add_exception_handler(UnicornException, unicorn_exception_handler) |
||||
|
``` |
||||
|
|
||||
|
This works the same as passing `exception_handlers` in the constructor. |
||||
|
|
||||
|
## Nested Router Precedence { #nested-router-precedence } |
||||
|
|
||||
|
When routers are nested (a router includes another router), exception handlers follow this precedence order (highest to lowest): |
||||
|
|
||||
|
1. The **innermost (child) router**'s handlers |
||||
|
2. The **parent router**'s handlers |
||||
|
3. The **app-level** handlers |
||||
|
|
||||
|
This means a child router can override its parent's handlers for its own routes. |
||||
|
|
||||
|
## Status Code Handlers { #status-code-handlers } |
||||
|
|
||||
|
Just like app-level handlers, you can use **integer status codes** as keys to handle specific HTTP error responses: |
||||
|
|
||||
|
```python |
||||
|
from starlette.exceptions import HTTPException |
||||
|
|
||||
|
def not_found_handler(request, exc): |
||||
|
return JSONResponse( |
||||
|
status_code=404, |
||||
|
content={"message": "Custom 404: resource not found"}, |
||||
|
) |
||||
|
|
||||
|
router = APIRouter( |
||||
|
exception_handlers={404: not_found_handler} |
||||
|
) |
||||
|
``` |
||||
|
|
||||
|
## Recap { #recap } |
||||
|
|
||||
|
* Pass `exception_handlers` to `APIRouter()` to scope handlers to that router's routes. |
||||
|
* Use `router.add_exception_handler()` to add handlers after creation. |
||||
|
* Router-level handlers **override** app-level handlers for the same exception type. |
||||
|
* Routes **outside** the router are unaffected. |
||||
|
* Nested routers follow **child > parent > app** precedence. |
||||
@ -0,0 +1,31 @@ |
|||||
|
from fastapi import APIRouter, FastAPI |
||||
|
from fastapi.responses import JSONResponse |
||||
|
|
||||
|
|
||||
|
class UnicornException(Exception): |
||||
|
def __init__(self, name: str): |
||||
|
self.name = name |
||||
|
|
||||
|
|
||||
|
router = APIRouter( |
||||
|
prefix="/unicorns", |
||||
|
exception_handlers={ |
||||
|
UnicornException: lambda request, exc: JSONResponse( |
||||
|
status_code=418, |
||||
|
content={ |
||||
|
"message": f"Oops! {exc.name} did something. There goes a rainbow..." |
||||
|
}, |
||||
|
) |
||||
|
}, |
||||
|
) |
||||
|
|
||||
|
|
||||
|
@router.get("/{name}") |
||||
|
async def read_unicorn(name: str): |
||||
|
if name == "yolo": |
||||
|
raise UnicornException(name=name) |
||||
|
return {"unicorn_name": name} |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
app.include_router(router) |
||||
@ -0,0 +1,52 @@ |
|||||
|
from fastapi import APIRouter, FastAPI, Request |
||||
|
from fastapi.responses import JSONResponse |
||||
|
|
||||
|
|
||||
|
class UnicornException(Exception): |
||||
|
def __init__(self, name: str): |
||||
|
self.name = name |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.exception_handler(UnicornException) |
||||
|
async def global_unicorn_exception_handler(request: Request, exc: UnicornException): |
||||
|
return JSONResponse( |
||||
|
status_code=418, |
||||
|
content={"message": f"Global handler: {exc.name} did something."}, |
||||
|
) |
||||
|
|
||||
|
|
||||
|
magic_router = APIRouter(prefix="/magic") |
||||
|
|
||||
|
|
||||
|
@magic_router.get("/unicorns/{name}") |
||||
|
async def read_magic_unicorn(name: str): |
||||
|
if name == "yolo": |
||||
|
raise UnicornException(name=name) |
||||
|
return {"unicorn_name": name} |
||||
|
|
||||
|
|
||||
|
def custom_unicorn_handler(request: Request, exc: UnicornException): |
||||
|
return JSONResponse( |
||||
|
status_code=418, |
||||
|
content={"message": f"Special handler: {exc.name} did something magical!"}, |
||||
|
) |
||||
|
|
||||
|
|
||||
|
special_router = APIRouter( |
||||
|
prefix="/special", |
||||
|
exception_handlers={UnicornException: custom_unicorn_handler}, |
||||
|
) |
||||
|
|
||||
|
|
||||
|
@special_router.get("/unicorns/{name}") |
||||
|
async def read_special_unicorn(name: str): |
||||
|
if name == "yolo": |
||||
|
raise UnicornException(name=name) |
||||
|
return {"unicorn_name": name} |
||||
|
|
||||
|
|
||||
|
app.include_router(magic_router) |
||||
|
app.include_router(special_router) |
||||
@ -0,0 +1,321 @@ |
|||||
|
from fastapi import APIRouter, FastAPI |
||||
|
from fastapi.responses import JSONResponse |
||||
|
from fastapi.testclient import TestClient |
||||
|
from fastapi.websockets import WebSocket |
||||
|
from starlette.requests import Request |
||||
|
|
||||
|
|
||||
|
class CustomExcA(Exception): |
||||
|
pass |
||||
|
|
||||
|
|
||||
|
class CustomExcB(Exception): |
||||
|
pass |
||||
|
|
||||
|
|
||||
|
def exc_a_handler(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "handled_a"}, status_code=400) |
||||
|
|
||||
|
|
||||
|
def exc_b_handler(request: Request, exc: CustomExcB) -> JSONResponse: |
||||
|
return JSONResponse({"error": "handled_b"}, status_code=400) |
||||
|
|
||||
|
|
||||
|
def test_basic_router_exception_handler(): |
||||
|
app = FastAPI() |
||||
|
router = APIRouter(exception_handlers={CustomExcA: exc_a_handler}) |
||||
|
|
||||
|
@router.get("/fail") |
||||
|
def fail(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
app.include_router(router) |
||||
|
client = TestClient(app) |
||||
|
resp = client.get("/fail") |
||||
|
assert resp.status_code == 400 |
||||
|
assert resp.json() == {"error": "handled_a"} |
||||
|
|
||||
|
|
||||
|
def test_isolation_between_routers(): |
||||
|
app = FastAPI() |
||||
|
router1 = APIRouter(prefix="/r1", exception_handlers={CustomExcA: exc_a_handler}) |
||||
|
router2 = APIRouter(prefix="/r2") |
||||
|
|
||||
|
@router1.get("/fail") |
||||
|
def fail1(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
@router2.get("/fail") |
||||
|
def fail2(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
app.include_router(router1) |
||||
|
app.include_router(router2) |
||||
|
client = TestClient(app, raise_server_exceptions=False) |
||||
|
|
||||
|
resp1 = client.get("/r1/fail") |
||||
|
assert resp1.status_code == 400 |
||||
|
assert resp1.json() == {"error": "handled_a"} |
||||
|
|
||||
|
resp2 = client.get("/r2/fail") |
||||
|
assert resp2.status_code == 500 |
||||
|
|
||||
|
|
||||
|
def test_router_overrides_app(): |
||||
|
def app_handler(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "app_handled"}, status_code=400) |
||||
|
|
||||
|
def router_handler(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "router_handled"}, status_code=400) |
||||
|
|
||||
|
app = FastAPI(exception_handlers={CustomExcA: app_handler}) |
||||
|
router = APIRouter(prefix="/r", exception_handlers={CustomExcA: router_handler}) |
||||
|
|
||||
|
@router.get("/fail") |
||||
|
def fail(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
@app.get("/app-fail") |
||||
|
def app_fail(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
app.include_router(router) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp = client.get("/r/fail") |
||||
|
assert resp.status_code == 400 |
||||
|
assert resp.json() == {"error": "router_handled"} |
||||
|
|
||||
|
resp_app = client.get("/app-fail") |
||||
|
assert resp_app.status_code == 400 |
||||
|
assert resp_app.json() == {"error": "app_handled"} |
||||
|
|
||||
|
|
||||
|
def test_app_fallback(): |
||||
|
def app_handler(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "app_handled"}, status_code=400) |
||||
|
|
||||
|
app = FastAPI(exception_handlers={CustomExcA: app_handler}) |
||||
|
router = APIRouter(prefix="/r") |
||||
|
|
||||
|
@router.get("/fail") |
||||
|
def fail(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
app.include_router(router) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp = client.get("/r/fail") |
||||
|
assert resp.status_code == 400 |
||||
|
assert resp.json() == {"error": "app_handled"} |
||||
|
|
||||
|
|
||||
|
def test_nested_router_precedence(): |
||||
|
def handler_a(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "A"}, status_code=400) |
||||
|
|
||||
|
def handler_b(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "B"}, status_code=400) |
||||
|
|
||||
|
router_a = APIRouter(prefix="/a", exception_handlers={CustomExcA: handler_a}) |
||||
|
router_b = APIRouter(prefix="/b", exception_handlers={CustomExcA: handler_b}) |
||||
|
|
||||
|
@router_a.get("/fail") |
||||
|
def fail_a(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
@router_b.get("/fail") |
||||
|
def fail_b(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
router_a.include_router(router_b) |
||||
|
app = FastAPI() |
||||
|
app.include_router(router_a) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp_b = client.get("/a/b/fail") |
||||
|
assert resp_b.status_code == 400 |
||||
|
assert resp_b.json() == {"error": "B"} |
||||
|
|
||||
|
resp_a = client.get("/a/fail") |
||||
|
assert resp_a.status_code == 400 |
||||
|
assert resp_a.json() == {"error": "A"} |
||||
|
|
||||
|
|
||||
|
def test_deep_nesting(): |
||||
|
def handler_a(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "A"}, status_code=400) |
||||
|
|
||||
|
def handler_b(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "B"}, status_code=400) |
||||
|
|
||||
|
def handler_c(request: Request, exc: CustomExcA) -> JSONResponse: |
||||
|
return JSONResponse({"error": "C"}, status_code=400) |
||||
|
|
||||
|
router_a = APIRouter(prefix="/a", exception_handlers={CustomExcA: handler_a}) |
||||
|
router_b = APIRouter(prefix="/b", exception_handlers={CustomExcA: handler_b}) |
||||
|
router_c = APIRouter(prefix="/c", exception_handlers={CustomExcA: handler_c}) |
||||
|
|
||||
|
@router_a.get("/fail") |
||||
|
def fail_a(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
@router_b.get("/fail") |
||||
|
def fail_b(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
@router_c.get("/fail") |
||||
|
def fail_c(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
router_b.include_router(router_c) |
||||
|
router_a.include_router(router_b) |
||||
|
app = FastAPI() |
||||
|
app.include_router(router_a) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp_c = client.get("/a/b/c/fail") |
||||
|
assert resp_c.status_code == 400 |
||||
|
assert resp_c.json() == {"error": "C"} |
||||
|
|
||||
|
resp_b = client.get("/a/b/fail") |
||||
|
assert resp_b.status_code == 400 |
||||
|
assert resp_b.json() == {"error": "B"} |
||||
|
|
||||
|
resp_a = client.get("/a/fail") |
||||
|
assert resp_a.status_code == 400 |
||||
|
assert resp_a.json() == {"error": "A"} |
||||
|
|
||||
|
|
||||
|
def test_add_exception_handler_method(): |
||||
|
app = FastAPI() |
||||
|
router = APIRouter() |
||||
|
router.add_exception_handler(CustomExcA, exc_a_handler) |
||||
|
|
||||
|
@router.get("/fail") |
||||
|
def fail(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
app.include_router(router) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp = client.get("/fail") |
||||
|
assert resp.status_code == 400 |
||||
|
assert resp.json() == {"error": "handled_a"} |
||||
|
|
||||
|
|
||||
|
def test_status_code_handler(): |
||||
|
from starlette.exceptions import HTTPException |
||||
|
|
||||
|
def not_found_handler(request: Request, exc: HTTPException) -> JSONResponse: |
||||
|
return JSONResponse({"error": "custom_404"}, status_code=404) |
||||
|
|
||||
|
app = FastAPI() |
||||
|
router = APIRouter(prefix="/r", exception_handlers={404: not_found_handler}) |
||||
|
|
||||
|
@router.get("/fail") |
||||
|
def fail(): |
||||
|
raise HTTPException(status_code=404, detail="not found") |
||||
|
|
||||
|
app.include_router(router) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp = client.get("/r/fail") |
||||
|
assert resp.status_code == 404 |
||||
|
assert resp.json() == {"error": "custom_404"} |
||||
|
|
||||
|
|
||||
|
def test_app_routes_unaffected(): |
||||
|
app = FastAPI() |
||||
|
router = APIRouter(prefix="/r", exception_handlers={CustomExcA: exc_a_handler}) |
||||
|
|
||||
|
@router.get("/fail") |
||||
|
def router_fail(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
@app.get("/fail") |
||||
|
def app_fail(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
app.include_router(router) |
||||
|
client = TestClient(app, raise_server_exceptions=False) |
||||
|
|
||||
|
resp_router = client.get("/r/fail") |
||||
|
assert resp_router.status_code == 400 |
||||
|
assert resp_router.json() == {"error": "handled_a"} |
||||
|
|
||||
|
resp_app = client.get("/fail") |
||||
|
assert resp_app.status_code == 500 |
||||
|
|
||||
|
|
||||
|
def test_multiple_exception_types(): |
||||
|
app = FastAPI() |
||||
|
router = APIRouter( |
||||
|
exception_handlers={ |
||||
|
CustomExcA: exc_a_handler, |
||||
|
CustomExcB: exc_b_handler, |
||||
|
} |
||||
|
) |
||||
|
|
||||
|
@router.get("/fail-a") |
||||
|
def fail_a(): |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
@router.get("/fail-b") |
||||
|
def fail_b(): |
||||
|
raise CustomExcB() |
||||
|
|
||||
|
app.include_router(router) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp_a = client.get("/fail-a") |
||||
|
assert resp_a.status_code == 400 |
||||
|
assert resp_a.json() == {"error": "handled_a"} |
||||
|
|
||||
|
resp_b = client.get("/fail-b") |
||||
|
assert resp_b.status_code == 400 |
||||
|
assert resp_b.json() == {"error": "handled_b"} |
||||
|
|
||||
|
|
||||
|
def test_websocket_router_exception_handler(): |
||||
|
app = FastAPI() |
||||
|
router = APIRouter( |
||||
|
exception_handlers={CustomExcA: exc_a_handler, 404: exc_a_handler} |
||||
|
) |
||||
|
|
||||
|
@router.websocket("/ws") |
||||
|
async def ws_endpoint(websocket: WebSocket): |
||||
|
await websocket.accept() |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
app.include_router(router) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
with client.websocket_connect("/ws") as ws: |
||||
|
# The exception handler returns a JSON response, but for websockets |
||||
|
# the connection should close with an error |
||||
|
try: |
||||
|
ws.receive_text() |
||||
|
except Exception: |
||||
|
pass |
||||
|
|
||||
|
|
||||
|
def test_websocket_nested_router_exception_handler(): |
||||
|
app = FastAPI() |
||||
|
router_a = APIRouter(prefix="/a", exception_handlers={CustomExcA: exc_a_handler}) |
||||
|
router_b = APIRouter(prefix="/b", exception_handlers={CustomExcA: exc_a_handler}) |
||||
|
|
||||
|
@router_b.websocket("/ws") |
||||
|
async def ws_endpoint(websocket: WebSocket): |
||||
|
await websocket.accept() |
||||
|
raise CustomExcA() |
||||
|
|
||||
|
router_a.include_router(router_b) |
||||
|
app.include_router(router_a) |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
with client.websocket_connect("/a/b/ws") as ws: |
||||
|
try: |
||||
|
ws.receive_text() |
||||
|
except Exception: |
||||
|
pass |
||||
@ -0,0 +1,101 @@ |
|||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
|
||||
|
from docs_src.handling_errors.tutorial007_py310 import app |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_get(): |
||||
|
response = client.get("/unicorns/shinny") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"unicorn_name": "shinny"} |
||||
|
|
||||
|
|
||||
|
def test_get_exception(): |
||||
|
response = client.get("/unicorns/yolo") |
||||
|
assert response.status_code == 418, response.text |
||||
|
assert response.json() == { |
||||
|
"message": "Oops! yolo did something. There goes a rainbow..." |
||||
|
} |
||||
|
|
||||
|
|
||||
|
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": { |
||||
|
"/unicorns/{name}": { |
||||
|
"get": { |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
}, |
||||
|
"422": { |
||||
|
"description": "Validation Error", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": { |
||||
|
"$ref": "#/components/schemas/HTTPValidationError" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
"summary": "Read Unicorn", |
||||
|
"operationId": "read_unicorn_unicorns__name__get", |
||||
|
"parameters": [ |
||||
|
{ |
||||
|
"required": True, |
||||
|
"schema": {"title": "Name", "type": "string"}, |
||||
|
"name": "name", |
||||
|
"in": "path", |
||||
|
} |
||||
|
], |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
"components": { |
||||
|
"schemas": { |
||||
|
"ValidationError": { |
||||
|
"title": "ValidationError", |
||||
|
"required": ["loc", "msg", "type"], |
||||
|
"type": "object", |
||||
|
"properties": { |
||||
|
"loc": { |
||||
|
"title": "Location", |
||||
|
"type": "array", |
||||
|
"items": { |
||||
|
"anyOf": [ |
||||
|
{"type": "string"}, |
||||
|
{"type": "integer"}, |
||||
|
] |
||||
|
}, |
||||
|
}, |
||||
|
"msg": {"title": "Message", "type": "string"}, |
||||
|
"type": {"title": "Error Type", "type": "string"}, |
||||
|
"input": {"title": "Input"}, |
||||
|
"ctx": {"title": "Context", "type": "object"}, |
||||
|
}, |
||||
|
}, |
||||
|
"HTTPValidationError": { |
||||
|
"title": "HTTPValidationError", |
||||
|
"type": "object", |
||||
|
"properties": { |
||||
|
"detail": { |
||||
|
"title": "Detail", |
||||
|
"type": "array", |
||||
|
"items": { |
||||
|
"$ref": "#/components/schemas/ValidationError" |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
@ -0,0 +1,143 @@ |
|||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
|
||||
|
from docs_src.handling_errors.tutorial008_py310 import app |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_get_magic(): |
||||
|
response = client.get("/magic/unicorns/shinny") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"unicorn_name": "shinny"} |
||||
|
|
||||
|
|
||||
|
def test_get_magic_exception(): |
||||
|
response = client.get("/magic/unicorns/yolo") |
||||
|
assert response.status_code == 418, response.text |
||||
|
assert response.json() == {"message": "Global handler: yolo did something."} |
||||
|
|
||||
|
|
||||
|
def test_get_special(): |
||||
|
response = client.get("/special/unicorns/shinny") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"unicorn_name": "shinny"} |
||||
|
|
||||
|
|
||||
|
def test_get_special_exception(): |
||||
|
response = client.get("/special/unicorns/yolo") |
||||
|
assert response.status_code == 418, response.text |
||||
|
assert response.json() == { |
||||
|
"message": "Special handler: yolo did something magical!" |
||||
|
} |
||||
|
|
||||
|
|
||||
|
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": { |
||||
|
"/magic/unicorns/{name}": { |
||||
|
"get": { |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
}, |
||||
|
"422": { |
||||
|
"description": "Validation Error", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": { |
||||
|
"$ref": "#/components/schemas/HTTPValidationError" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
"summary": "Read Magic Unicorn", |
||||
|
"operationId": "read_magic_unicorn_magic_unicorns__name__get", |
||||
|
"parameters": [ |
||||
|
{ |
||||
|
"required": True, |
||||
|
"schema": {"title": "Name", "type": "string"}, |
||||
|
"name": "name", |
||||
|
"in": "path", |
||||
|
} |
||||
|
], |
||||
|
} |
||||
|
}, |
||||
|
"/special/unicorns/{name}": { |
||||
|
"get": { |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
}, |
||||
|
"422": { |
||||
|
"description": "Validation Error", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": { |
||||
|
"$ref": "#/components/schemas/HTTPValidationError" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
"summary": "Read Special Unicorn", |
||||
|
"operationId": "read_special_unicorn_special_unicorns__name__get", |
||||
|
"parameters": [ |
||||
|
{ |
||||
|
"required": True, |
||||
|
"schema": {"title": "Name", "type": "string"}, |
||||
|
"name": "name", |
||||
|
"in": "path", |
||||
|
} |
||||
|
], |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
"components": { |
||||
|
"schemas": { |
||||
|
"ValidationError": { |
||||
|
"title": "ValidationError", |
||||
|
"required": ["loc", "msg", "type"], |
||||
|
"type": "object", |
||||
|
"properties": { |
||||
|
"loc": { |
||||
|
"title": "Location", |
||||
|
"type": "array", |
||||
|
"items": { |
||||
|
"anyOf": [ |
||||
|
{"type": "string"}, |
||||
|
{"type": "integer"}, |
||||
|
] |
||||
|
}, |
||||
|
}, |
||||
|
"msg": {"title": "Message", "type": "string"}, |
||||
|
"type": {"title": "Error Type", "type": "string"}, |
||||
|
"input": {"title": "Input"}, |
||||
|
"ctx": {"title": "Context", "type": "object"}, |
||||
|
}, |
||||
|
}, |
||||
|
"HTTPValidationError": { |
||||
|
"title": "HTTPValidationError", |
||||
|
"type": "object", |
||||
|
"properties": { |
||||
|
"detail": { |
||||
|
"title": "Detail", |
||||
|
"type": "array", |
||||
|
"items": { |
||||
|
"$ref": "#/components/schemas/ValidationError" |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
Loading…
Reference in new issue