Status Codes & Error Handling

from http import HTTPStatus

from fastapi import FastAPI, status
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()


class BookIn(BaseModel):
    title: str


@app.post("/books", status_code=status.HTTP_201_CREATED)
def create_book(book: BookIn) -> dict:
    return {"id": 1, "title": book.title}


@app.get("/teapot", status_code=HTTPStatus.IM_A_TEAPOT)
def teapot() -> dict:
    return {"short": "and stout"}


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int) -> None:
    return None

c = TestClient(app)
r = c.post("/books", json={"title": "Dune"})
print(r.status_code, r.json())
print(c.get("/teapot").status_code)
r = c.delete("/books/1")
print(r.status_code, repr(r.content))
spec = c.get("/openapi.json").json()["paths"]
print(list(spec["/books"]["post"]["responses"]))

201 {'id': 1, 'title': 'Dune'}
418
204 b''
['201', '422']

from fastapi import APIRouter, FastAPI, HTTPException, Response, status
from fastapi.testclient import TestClient
from pydantic import BaseModel

router = APIRouter(prefix="/books", tags=["books"])
BOOKS: dict[int, dict] = {}


class BookIn(BaseModel):
    title: str
    isbn: str


class BookOut(BookIn):
    id: int


@router.post("", response_model=BookOut, status_code=status.HTTP_201_CREATED)
def create_book(book: BookIn, response: Response):
    new_id = len(BOOKS) + 1
    BOOKS[new_id] = {"id": new_id, **book.model_dump()}
    response.headers["Location"] = f"/books/{new_id}"
    return BOOKS[new_id]


@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int) -> None:
    if BOOKS.pop(book_id, None) is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Book not found")


app = FastAPI()
app.include_router(router)

c = TestClient(app)
r = c.post("/books", json={"title": "Dune", "isbn": "978-0441013593"})
print(r.status_code, r.headers["location"], r.json())
r = c.delete("/books/1")
print(r.status_code, repr(r.content))
r = c.delete("/books/1")
print(r.status_code, r.json())

201 /books/1 {'title': 'Dune', 'isbn': '978-0441013593', 'id': 1}
204 b''
404 {'detail': 'Book not found'}

curl -i -X POST localhost:8000/books -H 'Content-Type: application/json' \
     -d '{"title": "Dune", "isbn": "978-0441013593"}'

HTTP/1.1 201 Created
location: /books/1
content-type: application/json

{"id":1,"title":"Dune","isbn":"978-0441013593"}

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException, status
from fastapi.testclient import TestClient

app = FastAPI()

BOOKS = {1: {"id": 1, "title": "Dune"}}


def get_book_or_404(book_id: int) -> dict:
    book = BOOKS.get(book_id)
    if book is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Book {book_id} not found",
        )
    return book


@app.get("/books/{book_id}")
def read_book(book: Annotated[dict, Depends(get_book_or_404)]) -> dict:
    return book


@app.get("/me")
def me(authorization: Annotated[str | None, Header()] = None) -> dict:
    if authorization != "Bearer demo":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {"user": "demo"}


@app.post("/books")
def create_book(title: str) -> dict:
    if any(b["title"] == title for b in BOOKS.values()):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail={"code": "duplicate_title", "title": title},
        )
    return {"title": title}

c = TestClient(app)
r = c.get("/books/1"); print(r.status_code, r.json())
r = c.get("/books/2"); print(r.status_code, r.json())
r = c.get("/me"); print(r.status_code, r.json(), r.headers["www-authenticate"])
r = c.post("/books?title=Dune"); print(r.status_code, r.json())

200 {'id': 1, 'title': 'Dune'}
404 {'detail': 'Book 2 not found'}
401 {'detail': 'Not authenticated'} Bearer
409 {'detail': {'code': 'duplicate_title', 'title': 'Dune'}}

curl -i localhost:8000/me
curl -i localhost:8000/books/2
curl -i -X DELETE localhost:8000/books/1

HTTP/1.1 401 Unauthorized
www-authenticate: Bearer
content-type: application/json

{"detail":"Not authenticated"}

HTTP/1.1 404 Not Found
content-type: application/json

{"detail":"Book 2 not found"}

HTTP/1.1 405 Method Not Allowed
allow: GET
content-type: application/json

{"detail":"Method Not Allowed"}

from typing import Annotated

from fastapi import APIRouter, Depends, FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from pydantic import BaseModel


# --- domain layer: knows nothing about HTTP -------------------------------
class BookshelfError(Exception):
    """Base class for domain errors."""


class Missing(BookshelfError):
    def __init__(self, kind: str, key: str):
        super().__init__(f"{kind} {key!r} not found")
        self.kind, self.key = kind, key


class Duplicate(BookshelfError):
    def __init__(self, kind: str, key: str):
        super().__init__(f"{kind} {key!r} already exists")
        self.kind, self.key = kind, key


class BookService:
    def __init__(self) -> None:
        self._books: dict[str, str] = {"978-0441013593": "Dune"}

    def get(self, isbn: str) -> str:
        try:
            return self._books[isbn]
        except KeyError:
            raise Missing("book", isbn) from None

    def create(self, isbn: str, title: str) -> str:
        if isbn in self._books:
            raise Duplicate("book", isbn)
        self._books[isbn] = title
        return title


_service = BookService()


def get_service() -> BookService:
    return _service


Service = Annotated[BookService, Depends(get_service)]


# --- web layer: maps domain errors to HTTP --------------------------------
class BookIn(BaseModel):
    isbn: str
    title: str


router = APIRouter(prefix="/books", tags=["books"])


@router.get("/{isbn}")
def read_book(isbn: str, service: Service) -> dict:
    return {"isbn": isbn, "title": service.get(isbn)}


@router.post("", status_code=status.HTTP_201_CREATED)
def create_book(book: BookIn, service: Service) -> dict:
    return {"isbn": book.isbn, "title": service.create(book.isbn, book.title)}


app = FastAPI()
app.include_router(router)


@app.exception_handler(Missing)
async def missing_handler(request: Request, exc: Missing) -> JSONResponse:
    return JSONResponse(status_code=status.HTTP_404_NOT_FOUND, content={"detail": str(exc)})


@app.exception_handler(Duplicate)
async def duplicate_handler(request: Request, exc: Duplicate) -> JSONResponse:
    return JSONResponse(status_code=status.HTTP_409_CONFLICT, content={"detail": str(exc)})


c = TestClient(app)
print(c.get("/books/978-0441013593").json())
r = c.get("/books/000"); print(r.status_code, r.json())
r = c.post("/books", json={"isbn": "1", "title": "A"}); print(r.status_code, r.json())
r = c.post("/books", json={"isbn": "1", "title": "A"}); print(r.status_code, r.json())

{'isbn': '978-0441013593', 'title': 'Dune'}
404 {'detail': "book '000' not found"}
201 {'isbn': '1', 'title': 'A'}
409 {'detail': "book '1' already exists"}

from fastapi import APIRouter, FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient


class BookshelfError(Exception):
    status_code = 400
    code = "bookshelf_error"


class Missing(BookshelfError):
    status_code = 404
    code = "not_found"


class Duplicate(BookshelfError):
    status_code = 409
    code = "duplicate"


app = FastAPI()


# one handler for the whole hierarchy: Starlette looks the handler up
# through the exception's class hierarchy (its MRO)
async def bookshelf_handler(request: Request, exc: BookshelfError) -> JSONResponse:
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.code, "detail": str(exc), "path": request.url.path},
    )


app.add_exception_handler(BookshelfError, bookshelf_handler)


@app.get("/missing")
def missing():
    raise Missing("no such book")


@app.get("/dup")
def dup():
    raise Duplicate("book exists")


@app.get("/other")
def other():
    raise BookshelfError("something else")


@app.get("/boom")
def boom():
    raise RuntimeError("bug")

c = TestClient(app, raise_server_exceptions=False)
for p in ("/missing", "/dup", "/other"):
    r = c.get(p); print(r.status_code, r.json())

404 {'code': 'not_found', 'detail': 'no such book', 'path': '/missing'}
409 {'code': 'duplicate', 'detail': 'book exists', 'path': '/dup'}
400 {'code': 'bookshelf_error', 'detail': 'something else', 'path': '/other'}

import logging
import uuid

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient

logging.basicConfig(level=logging.ERROR, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("bookshelf")

app = FastAPI()


@app.exception_handler(Exception)
async def unhandled(request: Request, exc: Exception) -> JSONResponse:
    error_id = uuid.uuid4().hex[:8]
    log.error("unhandled error %s on %s", error_id, request.url.path, exc_info=exc)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error", "error_id": error_id},
    )


@app.exception_handler(404)
async def not_found(request: Request, exc: Exception) -> JSONResponse:
    return JSONResponse(status_code=404, content={"detail": f"No route for {request.url.path}"})


@app.get("/boom")
def boom():
    raise RuntimeError("connection to db.internal failed")


# default TestClient re-raises server exceptions into the test
try:
    TestClient(app).get("/boom")
except RuntimeError as e:
    print("re-raised:", e)

c = TestClient(app, raise_server_exceptions=False)
r = c.get("/boom"); print(r.status_code, sorted(r.json()))
r = c.get("/nothing"); print(r.status_code, r.json())

re-raised: connection to db.internal failed
500 ['detail', 'error_id']
404 {'detail': 'No route for /nothing'}

from typing import Annotated

from fastapi import Body, FastAPI, Response, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()
SHELF: dict[str, dict] = {"978-0441013593": {"title": "Dune", "pages": 412}}


class Book(BaseModel):
    title: str
    pages: int


@app.put("/books/{isbn}")
def upsert(isbn: str, book: Annotated[Book, Body()]):
    if isbn in SHELF:
        SHELF[isbn] = book.model_dump()
        return SHELF[isbn]
    SHELF[isbn] = book.model_dump()
    return JSONResponse(status_code=status.HTTP_201_CREATED, content=jsonable_encoder(book))


c = TestClient(app)
body = {"title": "Emma", "pages": 474}
r = c.put("/books/1", json=body); print(r.status_code, r.json())
r = c.put("/books/1", json=body); print(r.status_code, r.json())
print(list(c.get("/openapi.json").json()["paths"]["/books/{isbn}"]["put"]["responses"]))

201 {'title': 'Emma', 'pages': 474}
200 {'title': 'Emma', 'pages': 474}
['200', '422']

from fastapi import FastAPI, Response, status
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()
SHELF: dict[str, dict] = {"978-0441013593": {"title": "Dune", "pages": 412}}


class Book(BaseModel):
    title: str
    pages: int


class BookOut(BaseModel):
    title: str


@app.put("/v2/books/{isbn}", response_model=BookOut)
def upsert_v2(isbn: str, book: Book, response: Response):
    if isbn not in SHELF:
        response.status_code = status.HTTP_201_CREATED
        response.headers["Location"] = f"/v2/books/{isbn}"
    SHELF[isbn] = book.model_dump()
    return SHELF[isbn]


c = TestClient(app)
body = {"title": "Emma", "pages": 474}
r = c.put("/v2/books/2", json=body); print(r.status_code, r.json(), r.headers["location"])
r = c.put("/v2/books/2", json=body); print(r.status_code, r.json(), "location" in r.headers)

201 {'title': 'Emma'} /v2/books/2
200 {'title': 'Emma'} False

import json
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Response, status
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()


class ErrorBody(BaseModel):
    detail: str


class Book(BaseModel):
    isbn: str
    title: str


BOOKS = {"1": Book(isbn="1", title="Dune")}

NOT_FOUND = {status.HTTP_404_NOT_FOUND: {"model": ErrorBody, "description": "Book not found"}}


@app.get(
    "/books/{isbn}",
    response_model=Book,
    responses={**NOT_FOUND, 409: {"model": ErrorBody}, 500: {"description": "Unexpected error"}},
)
def read_book(isbn: str):
    if isbn not in BOOKS:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Book not found")
    return BOOKS[isbn]


c = TestClient(app)
spec = c.get("/openapi.json").json()
print(json.dumps(spec["paths"]["/books/{isbn}"]["get"]["responses"], indent=1))

{
 "200": {
  "description": "Successful Response",
  "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Book"}}}
 },
 "404": {
  "description": "Book not found",
  "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ErrorBody"}}}
 },
 "409": {
  "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ErrorBody"}}},
  "description": "Conflict"
 },
 "500": {"description": "Unexpected error"},
 "422": {
  "description": "Validation Error",
  "content": {"application/json": {"schema": {"$ref": "#/components/schemas/HTTPValidationError"}}}
 }
}

from http import HTTPStatus
from typing import Any

from fastapi import FastAPI, HTTPException, Request
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict, Field
from starlette.exceptions import HTTPException as StarletteHTTPException

PROBLEM_JSON = "application/problem+json"


class Problem(BaseModel):
    """RFC 9457 problem details object (members used by Bookshelf)."""

    model_config = ConfigDict(extra="allow")

    type: str = Field(default="about:blank")
    title: str
    status: int
    detail: str | None = None
    instance: str | None = None


class ProblemResponse(JSONResponse):
    media_type = PROBLEM_JSON


class BookshelfProblem(Exception):
    """A domain error that carries its own problem type."""

    def __init__(self, status: int, type_: str, title: str, detail: str, **extensions: Any):
        self.status, self.type_, self.title, self.detail = status, type_, title, detail
        self.extensions = extensions


def problem(request: Request, status: int, *, type_: str = "about:blank",
            title: str | None = None, detail: str | None = None, **ext: Any) -> ProblemResponse:
    body = Problem(
        type=type_,
        title=title or HTTPStatus(status).phrase,
        status=status,
        detail=detail,
        instance=request.url.path,
        **ext,
    )
    return ProblemResponse(status_code=status, content=jsonable_encoder(body, exclude_none=True))


app = FastAPI(responses={"default": {"model": Problem, "content": {PROBLEM_JSON: {}}}})


@app.exception_handler(BookshelfProblem)
async def bookshelf_problem(request: Request, exc: BookshelfProblem):
    return problem(request, exc.status, type_=exc.type_, title=exc.title, detail=exc.detail, **exc.extensions)


@app.exception_handler(StarletteHTTPException)
async def http_problem(request: Request, exc: StarletteHTTPException):
    r = problem(request, exc.status_code, detail=str(exc.detail))
    r.headers.update(exc.headers or {})          # keep e.g. WWW-Authenticate
    return r


@app.exception_handler(RequestValidationError)
async def validation_problem(request: Request, exc: RequestValidationError):
    errors = [{"pointer": "/" + "/".join(map(str, e["loc"])), "detail": e["msg"]} for e in exc.errors()]
    return problem(request, 422, type_="https://api.example.com/problems/validation",
                   title="Request validation failed", detail="One or more fields are invalid.",
                   errors=errors)


@app.get("/books/{isbn}")
def read_book(isbn: str, limit: int = 10) -> dict:
    if isbn == "0":
        raise BookshelfProblem(409, "https://api.example.com/problems/out-of-stock",
                               "Out of stock", "No copies of this book are left.", isbn=isbn)
    if isbn == "1":
        raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})
    return {"isbn": isbn}


import json

c = TestClient(app)
for path in ("/books/0", "/books/1", "/nope", "/books/2?limit=x"):
    r = c.get(path)
    print(r.status_code, r.headers["content-type"], r.headers.get("www-authenticate"))
    print(json.dumps(r.json(), indent=1))

409 application/problem+json None
{"type": "https://api.example.com/problems/out-of-stock", "title": "Out of stock", "status": 409,
 "detail": "No copies of this book are left.", "instance": "/books/0", "isbn": "0"}
401 application/problem+json Bearer
{"type": "about:blank", "title": "Unauthorized", "status": 401,
 "detail": "Not authenticated", "instance": "/books/1"}
404 application/problem+json None
{"type": "about:blank", "title": "Not Found", "status": 404,
 "detail": "Not Found", "instance": "/nope"}
422 application/problem+json None
{"type": "https://api.example.com/problems/validation", "title": "Request validation failed", "status": 422,
 "detail": "One or more fields are invalid.", "instance": "/books/2",
 "errors": [{"pointer": "/query/limit", "detail": "Input should be a valid integer, unable to parse string as an integer"}]}

from typing import Annotated

from fastapi import Depends, FastAPI, Form
from fastapi.security import HTTPBearer
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()
bearer = HTTPBearer()


class Book(BaseModel):
    title: str


class BookOut(BaseModel):
    title: str
    pages: int


@app.post("/books")
def create(b: Book) -> Book:
    return b


@app.get("/shelf/")
def shelf() -> list: return []


@app.get("/secure")
def secure(cred: Annotated[object, Depends(bearer)]): return "ok"


@app.get("/bad", response_model=BookOut)
def bad(): return {"title": "x"}


@app.post("/login")
def login(username: Annotated[str, Form()]): return username


c = TestClient(app, raise_server_exceptions=False)
def show(label, r):
    print(label, r.status_code, str(r.json() if r.headers.get("content-type","").startswith("application/json") else r.text)[:150], r.headers.get("location"))

show("404 route", c.get("/nope"))
show("405", c.get("/books"))
show("307 slash", c.get("/shelf", follow_redirects=False))
show("422 no body", c.post("/books"))
show("422 bad json", c.post("/books", content="{bad", headers={"content-type": "application/json"}))
show("422 no ctype", c.post("/books", content='{"title":"x"}'))
show("422 wrong ctype", c.post("/books", content='{"title":"x"}', headers={"content-type": "text/plain"}))
show("401 bearer", c.get("/secure"))
show("500 resp validation", c.get("/bad"))
show("422 form", c.post("/login", data={}))

404 route 404 {'detail': 'Not Found'} None
405 405 {'detail': 'Method Not Allowed'} None
307 slash 307  http://testserver/shelf/
422 no body 422 {'detail': [{'type': 'missing', 'loc': ['body'], 'msg': 'Field required', 'input': None}]} None
422 bad json 422 {'detail': [{'type': 'json_invalid', 'loc': ['body', 1], 'msg': 'JSON decode error', 'input': {}, 'ctx': {'error': 'Expecting property name enclosed i None
422 no ctype 422 {'detail': [{'type': 'model_attributes_type', 'loc': ['body'], 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input':  None
422 wrong ctype 422 {'detail': [{'type': 'model_attributes_type', 'loc': ['body'], 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input':  None
401 bearer 401 {'detail': 'Not authenticated'} None
500 resp validation 500 Internal Server Error None
422 form 422 {'detail': [{'type': 'missing', 'loc': ['body', 'username'], 'msg': 'Field required', 'input': None}]} None