Project tutorials · 9 steps

Build a REST API in Python from scratch

Authentication, validation, pagination, idempotency and clean errors, in a working API built with the standard library so you can see how each one works.

What you'll build

Frameworks hide the parts of an API that are worth understanding. This tutorial builds a product reviews API with only the standard library: API keys and scopes, request validation, cursor pagination, safe retries and ownership checks. It's written as a class with one handle method that takes plain values, so you can test the whole thing without a network.

What you'll learn

Design a small API end to end: resources, routes, schemas, errors and auth
Combine authentication, scopes and object-level checks on every endpoint
Wire validation, persistence, cursor pagination and idempotency into one service
Test the whole API through a single entry point

What you need

Python classes and dictionaries, what HTTP status codes are, and a little SQL. It uses only the standard library. To serve it for real, call handle from a route in your web framework.

Step 1: Describe what a valid review is

The schema says what a review must contain, and validate checks a request against it. It returns every problem with its path, not just the first, so a client can fix them all at once. additionalProperties: False rejects unexpected fields, which catches typos in client code.

import base64, json, re, sqlite3

REVIEW_SCHEMA = {
    "type": "object",
    "required": ["rating", "text"],
    "properties": {
        "rating": {"type": "integer", "minimum": 1, "maximum": 5},
        "text": {"type": "string", "minLength": 1, "maxLength": 500},
    },
    "additionalProperties": False,
}


def validate(data, schema, path="$"):
    """Compact JSON Schema check (the keywords this API needs). Returns [(path, message)]."""
    kinds = {"object": dict, "string": str, "integer": int}
    t = schema.get("type")
    if t and (not isinstance(data, kinds[t]) or isinstance(data, bool)):
        return [(path, f"must be {t}")]
    errors = []
    if isinstance(data, int) and not isinstance(data, bool):
        if data < schema.get("minimum", data) or data > schema.get("maximum", data):
            errors.append((path, f"must be between {schema.get('minimum')} and {schema.get('maximum')}"))
    if isinstance(data, str) and not schema.get("minLength", 0) <= len(data) <= schema.get("maxLength", len(data)):
        errors.append((path, "has the wrong length"))
    if isinstance(data, dict):
        props = schema.get("properties", {})
        errors += [(f"{path}.{k}", "is required") for k in schema.get("required", []) if k not in data]
        for k, v in data.items():
            if k in props:
                errors += validate(v, props[k], f"{path}.{k}")
            elif schema.get("additionalProperties") is False:
                errors.append((f"{path}.{k}", "is not allowed"))
    return errors

Step 2: Use one error shape everywhere

Every error from this API looks the same: a type, a message and sometimes details. Clients can then handle errors in one place. Keeping that in a small helper is what keeps it consistent.

TYPES = {400: "invalid_request_error", 401: "authentication_error", 403: "permission_error",
         404: "not_found_error", 422: "invalid_request_error"}


def err(status, message, **extra):
    return status, {}, {"error": {"type": TYPES.get(status, "api_error"), "message": message, **extra}}

Step 3: Make pagination cursors opaque

A cursor marks where the last page ended. Encoding it means clients treat it as a token and don't build it themselves, so you can change what's inside later. Cursors also stay correct when rows are added, where page numbers would skip or repeat them.

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


def decode_cursor(cursor):
    return int(json.loads(base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4)))["after"])

Rather write it yourself?

In the lesson you write this from a starter, in your browser with nothing to install, and checks tell you what's wrong as you go. A free account opens it and all 244 lessons.

Create a free account

Step 4: Set up storage and routing

handle matches a method and a path to a function. A real framework does this for you, and this is roughly what it does. Lowercasing the headers matters because HTTP header names aren't case-sensitive. The wrong method on a real route gets a 405, and an unknown route gets a 404.

class ReviewsAPI:
    def __init__(self, db, keys, products):
        self.db, self.keys, self.products = db, keys, set(products)
        self.idempotent = {}
        db.execute("""CREATE TABLE IF NOT EXISTS reviews (
            id INTEGER PRIMARY KEY, sku TEXT NOT NULL, rating INTEGER NOT NULL,
            text TEXT NOT NULL, author TEXT NOT NULL)""")

    def handle(self, method, path, headers=None, body=None, query=None):
        headers = {k.lower(): v for k, v in (headers or {}).items()}
        if m := re.fullmatch(r"/products/([\w-]+)/reviews", path):
            if method == "POST":
                return self.create_review(m.group(1), headers, body)
            if method == "GET":
                return self.list_reviews(m.group(1), headers, query or {})
            return err(405, "method not allowed")
        if m := re.fullmatch(r"/reviews/(\d+)", path):
            if method == "DELETE":
                return self.delete_review(int(m.group(1)), headers)
            return err(405, "method not allowed")
        return err(404, "no such route")

Step 5: Check who's calling

_row fetches one review as a dictionary, and _auth checks the caller. A missing or unknown key gets a 401: we don't know who you are. A known key without the right scope gets a 403: we know, and you can't do that. Getting that distinction right saves your API's users a lot of confused debugging.

    def _row(self, review_id):
        r = self.db.execute("SELECT id, sku, rating, text, author FROM reviews WHERE id = ?", (review_id,)).fetchone()
        return dict(zip(["id", "sku", "rating", "text", "author"], r)) if r else None

    def _auth(self, headers, scope):
        value = headers.get("authorization", "")
        if not value.startswith("Bearer "):
            return err(401, "send Authorization: Bearer <key>")
        record = self.keys.get(value[len("Bearer "):].strip())
        if record is None:
            return err(401, "unknown API key")
        if scope not in record["scopes"]:
            return err(403, f"this key needs the {scope} scope")
        return record

Step 6: Create a review, safely

Validation returns 422 with every problem listed. The idempotency key solves a real problem: if a client's connection drops, it can't tell whether the request worked. Retrying with the same key returns the first response instead of creating a duplicate. It's stored per owner, so one user's key can never reveal another's response.

    def create_review(self, sku, headers, body):
        who = self._auth(headers, "reviews:write")
        if isinstance(who, tuple):
            return who
        idem = headers.get("idempotency-key")
        if idem and (who["owner"], idem) in self.idempotent:
            return self.idempotent[(who["owner"], idem)]
        if sku not in self.products:
            return err(404, f"no product {sku}")
        problems = validate(body, REVIEW_SCHEMA)
        if problems:
            return err(422, "the review isn't valid", details=[{"path": p, "message": m} for p, m in problems])
        with self.db:
            cur = self.db.execute(
                "INSERT INTO reviews (sku, rating, text, author) VALUES (?, ?, ?, ?)",
                (sku, body["rating"], body["text"], who["owner"]),
            )
        response = (201, {}, self._row(cur.lastrowid))
        if idem:
            self.idempotent[(who["owner"], idem)] = response
        return response

Step 7: List reviews with pagination

It asks for one row more than the page size, which is a cheap way to find out whether another page exists. limit is bounded, so no client can ask for a million rows. Filtering with id > cursor stays fast however deep the page, where OFFSET gets slower.

    def list_reviews(self, sku, headers, query):
        who = self._auth(headers, "reviews:read")
        if isinstance(who, tuple):
            return who
        if sku not in self.products:
            return err(404, f"no product {sku}")
        try:
            limit = int(query.get("limit", 10))
        except ValueError:
            return err(400, "limit must be a whole number")
        if not 1 <= limit <= 50:
            return err(400, "limit must be between 1 and 50")
        after = 0
        if "cursor" in query:
            try:
                after = decode_cursor(query["cursor"])
            except Exception:
                return err(400, "invalid cursor")
        rows = self.db.execute(
            "SELECT id FROM reviews WHERE sku = ? AND id > ? ORDER BY id LIMIT ?", (sku, after, limit + 1)
        ).fetchall()
        page = [self._row(r[0]) for r in rows[:limit]]
        has_more = len(rows) > limit
        return 200, {}, {"data": page, "has_more": has_more, "next_cursor": encode_cursor(page[-1]["id"]) if has_more else None}

Step 8: Let people delete only their own

Having a write scope isn't enough, because the review must also belong to the caller. Checking only the scope is one of the most common API security bugs. Someone else's review returns the same 404 as one that doesn't exist, so the API doesn't reveal which ids are real.

    def delete_review(self, review_id, headers):
        who = self._auth(headers, "reviews:write")
        if isinstance(who, tuple):
            return who
        review = self._row(review_id)
        if review is None or review["author"] != who["owner"]:
            return err(404, f"no review {review_id}")
        with self.db:
            self.db.execute("DELETE FROM reviews WHERE id = ?", (review_id,))
        return 204, {}, None

Step 9: Try it

Each response is a status, headers and a body. Because handle takes plain values, tests can call it directly, and they run in milliseconds.

keys = {"k_ada": {"owner": "ada", "scopes": ["reviews:read", "reviews:write"]}}
api = ReviewsAPI(sqlite3.connect(":memory:"), keys, ["TENT-2", "MUG-1"])
auth = {"Authorization": "Bearer k_ada"}
print(api.handle("POST", "/products/TENT-2/reviews", auth, {"rating": 5, "text": "Kept us dry all week."}))
print(api.handle("GET", "/products/TENT-2/reviews", auth, query={"limit": "5"}))

What it prints

(201, {}, {'id': 1, 'sku': 'TENT-2', 'rating': 5, 'text': 'Kept us dry all week.', 'author': 'ada'})
(200, {}, {'data': [{'id': 1, 'sku': 'TENT-2', 'rating': 5, 'text': 'Kept us dry all week.', 'author': 'ada'}], 'has_more': False, 'next_cursor': None})

Where to take it next

Serve it: call handle from a route in Flask or FastAPI, or from http.server.
Keep idempotency keys in the database, so they survive a restart, and let them expire.
Hash API keys rather than storing them as written.
Add rate limits, and write an OpenAPI description.

Mistakes to watch for

Validating input before checking whether the caller may use the endpoint at all.
Answering 403 for other users' objects, which confirms they exist.
Sharing idempotency keys across clients.
Counting all rows to decide has_more.

The complete program

import base64, json, re, sqlite3

REVIEW_SCHEMA = {
    "type": "object",
    "required": ["rating", "text"],
    "properties": {
        "rating": {"type": "integer", "minimum": 1, "maximum": 5},
        "text": {"type": "string", "minLength": 1, "maxLength": 500},
    },
    "additionalProperties": False,
}


def validate(data, schema, path="$"):
    """Compact JSON Schema check (the keywords this API needs). Returns [(path, message)]."""
    kinds = {"object": dict, "string": str, "integer": int}
    t = schema.get("type")
    if t and (not isinstance(data, kinds[t]) or isinstance(data, bool)):
        return [(path, f"must be {t}")]
    errors = []
    if isinstance(data, int) and not isinstance(data, bool):
        if data < schema.get("minimum", data) or data > schema.get("maximum", data):
            errors.append((path, f"must be between {schema.get('minimum')} and {schema.get('maximum')}"))
    if isinstance(data, str) and not schema.get("minLength", 0) <= len(data) <= schema.get("maxLength", len(data)):
        errors.append((path, "has the wrong length"))
    if isinstance(data, dict):
        props = schema.get("properties", {})
        errors += [(f"{path}.{k}", "is required") for k in schema.get("required", []) if k not in data]
        for k, v in data.items():
            if k in props:
                errors += validate(v, props[k], f"{path}.{k}")
            elif schema.get("additionalProperties") is False:
                errors.append((f"{path}.{k}", "is not allowed"))
    return errors


TYPES = {400: "invalid_request_error", 401: "authentication_error", 403: "permission_error",
         404: "not_found_error", 422: "invalid_request_error"}


def err(status, message, **extra):
    return status, {}, {"error": {"type": TYPES.get(status, "api_error"), "message": message, **extra}}


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


def decode_cursor(cursor):
    return int(json.loads(base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4)))["after"])


class ReviewsAPI:
    def __init__(self, db, keys, products):
        self.db, self.keys, self.products = db, keys, set(products)
        self.idempotent = {}
        db.execute("""CREATE TABLE IF NOT EXISTS reviews (
            id INTEGER PRIMARY KEY, sku TEXT NOT NULL, rating INTEGER NOT NULL,
            text TEXT NOT NULL, author TEXT NOT NULL)""")

    def handle(self, method, path, headers=None, body=None, query=None):
        headers = {k.lower(): v for k, v in (headers or {}).items()}
        if m := re.fullmatch(r"/products/([\w-]+)/reviews", path):
            if method == "POST":
                return self.create_review(m.group(1), headers, body)
            if method == "GET":
                return self.list_reviews(m.group(1), headers, query or {})
            return err(405, "method not allowed")
        if m := re.fullmatch(r"/reviews/(\d+)", path):
            if method == "DELETE":
                return self.delete_review(int(m.group(1)), headers)
            return err(405, "method not allowed")
        return err(404, "no such route")

    def _row(self, review_id):
        r = self.db.execute("SELECT id, sku, rating, text, author FROM reviews WHERE id = ?", (review_id,)).fetchone()
        return dict(zip(["id", "sku", "rating", "text", "author"], r)) if r else None

    def _auth(self, headers, scope):
        value = headers.get("authorization", "")
        if not value.startswith("Bearer "):
            return err(401, "send Authorization: Bearer <key>")
        record = self.keys.get(value[len("Bearer "):].strip())
        if record is None:
            return err(401, "unknown API key")
        if scope not in record["scopes"]:
            return err(403, f"this key needs the {scope} scope")
        return record

    def create_review(self, sku, headers, body):
        who = self._auth(headers, "reviews:write")
        if isinstance(who, tuple):
            return who
        idem = headers.get("idempotency-key")
        if idem and (who["owner"], idem) in self.idempotent:
            return self.idempotent[(who["owner"], idem)]
        if sku not in self.products:
            return err(404, f"no product {sku}")
        problems = validate(body, REVIEW_SCHEMA)
        if problems:
            return err(422, "the review isn't valid", details=[{"path": p, "message": m} for p, m in problems])
        with self.db:
            cur = self.db.execute(
                "INSERT INTO reviews (sku, rating, text, author) VALUES (?, ?, ?, ?)",
                (sku, body["rating"], body["text"], who["owner"]),
            )
        response = (201, {}, self._row(cur.lastrowid))
        if idem:
            self.idempotent[(who["owner"], idem)] = response
        return response

    def list_reviews(self, sku, headers, query):
        who = self._auth(headers, "reviews:read")
        if isinstance(who, tuple):
            return who
        if sku not in self.products:
            return err(404, f"no product {sku}")
        try:
            limit = int(query.get("limit", 10))
        except ValueError:
            return err(400, "limit must be a whole number")
        if not 1 <= limit <= 50:
            return err(400, "limit must be between 1 and 50")
        after = 0
        if "cursor" in query:
            try:
                after = decode_cursor(query["cursor"])
            except Exception:
                return err(400, "invalid cursor")
        rows = self.db.execute(
            "SELECT id FROM reviews WHERE sku = ? AND id > ? ORDER BY id LIMIT ?", (sku, after, limit + 1)
        ).fetchall()
        page = [self._row(r[0]) for r in rows[:limit]]
        has_more = len(rows) > limit
        return 200, {}, {"data": page, "has_more": has_more, "next_cursor": encode_cursor(page[-1]["id"]) if has_more else None}

    def delete_review(self, review_id, headers):
        who = self._auth(headers, "reviews:write")
        if isinstance(who, tuple):
            return who
        review = self._row(review_id)
        if review is None or review["author"] != who["owner"]:
            return err(404, f"no review {review_id}")
        with self.db:
            self.db.execute("DELETE FROM reviews WHERE id = ?", (review_id,))
        return 204, {}, None


keys = {"k_ada": {"owner": "ada", "scopes": ["reviews:read", "reviews:write"]}}
api = ReviewsAPI(sqlite3.connect(":memory:"), keys, ["TENT-2", "MUG-1"])
auth = {"Authorization": "Bearer k_ada"}
print(api.handle("POST", "/products/TENT-2/reviews", auth, {"rating": 5, "text": "Kept us dry all week."}))
print(api.handle("GET", "/products/TENT-2/reviews", auth, query={"limit": "5"}))

Write it yourself

In the lesson you write this from a starter, in your browser with nothing to install, and checks tell you what's wrong as you go. A free account opens it and all 244 lessons.

Create a free account