Building REST APIs

import itertools
from threading import Lock

from flask import Flask, Blueprint, jsonify, request, url_for
from flask.views import MethodView
from werkzeug.exceptions import HTTPException


class ApiError(Exception):
    """An error that becomes an RFC 9457 Problem Details response."""

    def __init__(self, status, title, detail=None, **extra):
        super().__init__(title)
        self.status, self.title, self.detail, self.extra = status, title, detail, extra


def problem(status, title, detail=None, **extra):
    body = {"type": "about:blank", "title": title, "status": status}
    if detail:
        body["detail"] = detail
    body.update(extra)
    response = jsonify(body)
    response.status_code = status
    response.mimetype = "application/problem+json"
    return response


def register_error_handlers(app):
    @app.errorhandler(ApiError)
    def handle_api_error(exc):
        return problem(exc.status, exc.title, exc.detail, **exc.extra)

    @app.errorhandler(HTTPException)
    def handle_http_exception(exc):
        response = problem(exc.code, exc.name, exc.description)
        if getattr(exc, "valid_methods", None):
            response.headers["Allow"] = ", ".join(exc.valid_methods)
        return response

import math

SORTABLE = {"id", "title", "author", "year"}
MAX_PER_PAGE = 100


def int_arg(args, name, default, minimum, maximum=None):
    raw = args.get(name)
    if raw is None:
        return default
    try:
        value = int(raw)
    except ValueError:
        raise ApiError(400, "Invalid query parameter", f"{name} must be an integer", parameter=name)
    if value < minimum or (maximum is not None and value > maximum):
        raise ApiError(400, "Invalid query parameter", f"{name} is out of range", parameter=name)
    return value


def query_books(books, args):
    """Filter, sort and paginate a list of book dictionaries; returns (items, meta)."""
    page = int_arg(args, "page", 1, 1)
    per_page = int_arg(args, "per_page", 20, 1, MAX_PER_PAGE)
    if author := args.get("author"):
        books = [b for b in books if b["author"].casefold() == author.casefold()]
    if q := args.get("q"):
        books = [b for b in books if q.casefold() in b["title"].casefold()]
    for key in reversed(args.get("sort", "id").split(",")):  # stable sort: last key first
        field = key.lstrip("-")
        if field not in SORTABLE:
            raise ApiError(400, "Invalid query parameter", f"cannot sort by {field}", parameter="sort")
        books = sorted(books, key=lambda b: (b.get(field) is None, b.get(field) or 0),
                       reverse=key.startswith("-"))
    total = len(books)
    pages = max(1, math.ceil(total / per_page))
    start = (page - 1) * per_page
    meta = {"page": page, "per_page": per_page, "total": total, "pages": pages}
    return books[start:start + per_page], meta


def link_header(endpoint, meta, args):
    """RFC 8288 Link header with first, last, prev and next relations."""
    def link(page, rel):
        return f'<{url_for(endpoint, **{**args.to_dict(), "page": page})}>; rel="{rel}"'

    links = [link(1, "first"), link(meta["pages"], "last")]
    if meta["page"] > 1:
        links.append(link(meta["page"] - 1, "prev"))
    if meta["page"] < meta["pages"]:
        links.append(link(meta["page"] + 1, "next"))
    return ", ".join(links)

import base64
import json


def encode_cursor(book_id):
    return base64.urlsafe_b64encode(json.dumps({"after": book_id}).encode()).decode()


def decode_cursor(token):
    try:
        return int(json.loads(base64.urlsafe_b64decode(token.encode()))["after"])
    except (ValueError, KeyError, TypeError):
        raise ApiError(400, "Invalid cursor", "the cursor is malformed", parameter="after")


def keyset_page(books, after=None, limit=20):
    last_seen = decode_cursor(after) if after else 0
    window = [b for b in sorted(books, key=lambda b: b["id"]) if b["id"] > last_seen][:limit + 1]
    items = window[:limit]
    next_cursor = encode_cursor(items[-1]["id"]) if len(window) > limit else None
    return items, next_cursor

class BookRepository:
    """In-memory stand-in for the database layer of the earlier pages."""

    def __init__(self):
        self._books, self._ids, self._lock = {}, itertools.count(1), Lock()

    def all(self):
        return list(self._books.values())

    def get(self, book_id):
        return self._books.get(book_id)

    def find_isbn(self, isbn):
        return next((b for b in self._books.values() if b["isbn"] == isbn), None)

    def add(self, data):
        with self._lock:
            book = {"id": next(self._ids), **data}
            self._books[book["id"]] = book
            return book

    def replace(self, book_id, data):
        self._books[book_id] = {"id": book_id, **data}
        return self._books[book_id]

    def delete(self, book_id):
        return self._books.pop(book_id, None)


FIELDS = {"title": str, "author": str, "isbn": str, "year": int}
REQUIRED = {"title", "author", "isbn"}


def parse_book(payload, partial=False):
    """Hand-written validation; the real projects below use a library."""
    if not isinstance(payload, dict):
        raise ApiError(422, "Validation failed", errors={"_body": ["Expected a JSON object."]})
    errors = {name: ["Unknown field."] for name in payload.keys() - FIELDS.keys()}
    if not partial:
        for name in REQUIRED - payload.keys():
            errors[name] = ["Missing data for required field."]
    for name, kind in FIELDS.items():
        if name in payload:
            value = payload[name]
            if kind is int:
                valid = isinstance(value, int) and not isinstance(value, bool)
            else:
                valid = isinstance(value, str) and value.strip() != ""
            if not valid:
                errors[name] = ["Must be an integer." if kind is int else "Must be a non-empty string."]
    if errors:
        raise ApiError(422, "Validation failed", "The request body is not valid.", errors=errors)
    return {name: payload[name] for name in FIELDS if name in payload}

from werkzeug.http import generate_etag


def book_resource(book):
    """Representation with hypermedia links (a step towards Richardson level 3)."""
    return {**book, "links": {"self": url_for("api_v1.book_item", book_id=book["id"]),
                              "collection": url_for("api_v1.book_collection")}}


def etag_for(book):
    return generate_etag(json.dumps(book, sort_keys=True).encode())


class BookCollection(MethodView):
    def __init__(self, repo, parse):
        self.repo, self.parse = repo, parse

    def get(self):
        items, meta = query_books(self.repo.all(), request.args)
        response = jsonify({"items": [book_resource(b) for b in items], "meta": meta})
        response.headers["Link"] = link_header("api_v1.book_collection", meta, request.args)
        response.headers["X-Total-Count"] = str(meta["total"])
        return response

    def post(self):
        data = self.parse(request.get_json())          # 400 or 415 here, 422 in parse
        if self.repo.find_isbn(data["isbn"]):
            raise ApiError(409, "Duplicate ISBN", f"A book with ISBN {data['isbn']} already exists.")
        book = self.repo.add(data)
        location = url_for("api_v1.book_item", book_id=book["id"])
        return jsonify(book_resource(book)), 201, {"Location": location}


class BookItem(MethodView):
    def __init__(self, repo, parse):
        self.repo, self.parse = repo, parse

    def _book_or_404(self, book_id):
        book = self.repo.get(book_id)
        if book is None:
            raise ApiError(404, "Book not found", f"There is no book with id {book_id}.")
        return book

    def _check_isbn(self, isbn, own_id):
        other = self.repo.find_isbn(isbn)
        if other and other["id"] != own_id:
            raise ApiError(409, "Duplicate ISBN", f"A book with ISBN {isbn} already exists.")

    def _check_if_match(self, book):
        if request.if_match and not request.if_match.contains(etag_for(book)):
            raise ApiError(412, "Precondition failed", "The book changed since you read it.")

    def get(self, book_id):
        book = self._book_or_404(book_id)
        response = jsonify(book_resource(book))
        response.set_etag(etag_for(book))
        return response.make_conditional(request)       # 304 when If-None-Match matches

    def put(self, book_id):                              # replace: all required fields again
        book = self._book_or_404(book_id)
        self._check_if_match(book)
        data = self.parse(request.get_json())
        self._check_isbn(data["isbn"], book_id)
        return jsonify(book_resource(self.repo.replace(book_id, data)))

    def patch(self, book_id):                            # JSON Merge Patch (RFC 7396)
        book = self._book_or_404(book_id)
        self._check_if_match(book)
        changes = self.parse(request.get_json(), partial=True)
        if "isbn" in changes:
            self._check_isbn(changes["isbn"], book_id)
        merged = {**{k: v for k, v in book.items() if k != "id"}, **changes}
        return jsonify(book_resource(self.repo.replace(book_id, merged)))

    def delete(self, book_id):
        self._book_or_404(book_id)
        self.repo.delete(book_id)
        return "", 204


def create_app(repo=None, parse=parse_book):
    app = Flask(__name__)
    repo = repo if repo is not None else BookRepository()
    api = Blueprint("api_v1", __name__, url_prefix="/api/v1")
    api.add_url_rule("/books", view_func=BookCollection.as_view("book_collection", repo, parse))
    api.add_url_rule("/books/<int:book_id>", view_func=BookItem.as_view("book_item", repo, parse))
    app.register_blueprint(api)
    register_error_handlers(app)
    return app

repo = BookRepository()
app = create_app(repo)
client = app.test_client()

created = client.post("/api/v1/books", json={"title": "Dune", "author": "Frank Herbert",
                                              "isbn": "9780000000001", "year": 1965})
assert created.status_code == 201
location = created.headers["Location"]
assert location == "/api/v1/books/1"

for i, (title, author) in enumerate([("Emma", "Jane Austen"), ("Persuasion", "Jane Austen"),
                                      ("Dune Messiah", "Frank Herbert")], start=2):
    client.post("/api/v1/books", json={"title": title, "author": author, "isbn": f"978000000000{i}"})

assert client.post("/api/v1/books", json={"title": "Dune", "author": "F", "isbn": "9780000000001"}).status_code == 409
invalid = client.post("/api/v1/books", json={"title": "", "titel": "typo"})
assert invalid.status_code == 422 and invalid.mimetype == "application/problem+json"
assert client.post("/api/v1/books", data="{broken", content_type="application/json").status_code == 400
assert client.post("/api/v1/books", data="title=Dune", content_type="text/plain").status_code == 415
assert client.get("/api/v1/books/99").status_code == 404
assert client.patch("/api/v1/books").status_code == 405

item = client.get(location)
etag = item.headers["ETag"]
assert client.get(location, headers={"If-None-Match": etag}).status_code == 304

patched = client.patch(location, data=json.dumps({"year": 1966}), content_type="application/merge-patch+json")
assert patched.json["year"] == 1966 and patched.json["title"] == "Dune"
assert client.put(location, json={"title": "Dune"}, headers={"If-Match": etag}).status_code == 412
assert client.put(location, json={"title": "Dune", "author": "Frank Herbert", "isbn": "9780000000001"}).status_code == 200
assert "year" not in client.get(location).json            # PUT replaced the whole book

page = client.get("/api/v1/books?author=jane austen&sort=-title&per_page=1&page=2")
assert [b["title"] for b in page.json["items"]] == ["Emma"]
assert page.headers["X-Total-Count"] == "2"
assert client.get("/api/v1/books?per_page=1000").status_code == 400
assert client.get("/api/v1/books?sort=password").status_code == 400
assert client.delete(location).status_code == 204 and client.get(location).status_code == 404

curl -i -X POST http://localhost:5000/api/v1/books \
     -H 'Content-Type: application/json' \
     -d '{"title": "Dune", "author": "Frank Herbert", "isbn": "9780000000001", "year": 1965}'

HTTP/1.1 201 CREATED
Content-Type: application/json
Location: /api/v1/books/1

{"author":"Frank Herbert","id":1,"isbn":"9780000000001","links":{"collection":"/api/v1/books","self":"/api/v1/books/1"},"title":"Dune","year":1965}

curl -i -X POST http://localhost:5000/api/v1/books \
     -H 'Content-Type: application/json' -d '{"title": "", "titel": "typo"}'

HTTP/1.1 422 UNPROCESSABLE ENTITY
Content-Type: application/problem+json

{"detail":"The request body is not valid.","errors":{"author":["Missing data for required field."],"isbn":["Missing data for required field."],"titel":["Unknown field."],"title":["Must be a non-empty string."]},"status":422,"title":"Validation failed","type":"about:blank"}

import datetime

from werkzeug.http import http_date

DEPRECATED_SINCE = datetime.datetime(2026, 10, 1, tzinfo=datetime.timezone.utc)
SUNSET = datetime.datetime(2027, 6, 30, tzinfo=datetime.timezone.utc)
BOOKS = {1: {"id": 1, "title": "Dune", "authors": ["Frank Herbert"]}}


def v1_book(book):                      # the old shape: one author string
    return {"id": book["id"], "title": book["title"], "author": book["authors"][0]}


def v2_book(book):                      # the new shape: a list of authors
    return {"id": book["id"], "title": book["title"], "authors": book["authors"]}


v1 = Blueprint("v1", __name__, url_prefix="/api/v1")
v2 = Blueprint("v2", __name__, url_prefix="/api/v2")


@v1.get("/books/<int:book_id>")
def v1_get(book_id):
    return jsonify(v1_book(BOOKS[book_id]))


@v2.get("/books/<int:book_id>")
def v2_get(book_id):
    return jsonify(v2_book(BOOKS[book_id]))


@v1.after_request
def mark_v1_deprecated(response):
    response.headers["Deprecation"] = f"@{int(DEPRECATED_SINCE.timestamp())}"
    response.headers["Sunset"] = http_date(SUNSET)
    response.headers.add("Link", f'<{request.path.replace("/v1/", "/v2/", 1)}>; rel="successor-version"')
    return response


negotiated = Flask(__name__)
negotiated.register_blueprint(v1)
negotiated.register_blueprint(v2)


@negotiated.get("/api/books/<int:book_id>")           # version chosen by the Accept header
def negotiated_get(book_id):
    media = request.accept_mimetypes.best_match(
        ["application/vnd.bookshelf.v2+json", "application/vnd.bookshelf.v1+json"])
    if media is None:
        raise ApiError(406, "Not acceptable", "Supported: application/vnd.bookshelf.v1+json and v2+json")
    shape = v1_book if media.endswith("v1+json") else v2_book
    response = jsonify(shape(BOOKS[book_id]))
    response.mimetype = media
    response.headers["Vary"] = "Accept"
    return response


register_error_handlers(negotiated)
vc = negotiated.test_client()
old = vc.get("/api/v1/books/1")
assert old.json == {"id": 1, "title": "Dune", "author": "Frank Herbert"}
assert old.headers["Deprecation"] == "@1790812800" and old.headers["Sunset"] == "Wed, 30 Jun 2027 00:00:00 GMT"
assert old.headers["Link"] == '</api/v2/books/1>; rel="successor-version"'
assert "Deprecation" not in vc.get("/api/v2/books/1").headers
assert vc.get("/api/books/1", headers={"Accept": "application/vnd.bookshelf.v2+json"}).json["authors"]
assert vc.get("/api/books/1", headers={"Accept": "text/html"}).status_code == 406