Flask — Building REST APIs

1. Overview

Flask remains one of the most pragmatic frameworks for building REST APIs in Python. It is unopinionated, composable, and has aged well — Flask 3.x on Python 3.11+ with type hints, Pydantic v2 validation, Marshmallow schemas, and Flask-Smorest for OpenAPI 3 generation produces a stack that is both ergonomic and production-grade. A well-structured Flask REST API in 2026 looks like this:

• Application factory (create_app()) returning a configured Flask instance — required for testability and multi-config deployments.
• Blueprints organized by resource and mounted under versioned prefixes (/api/v1/users, /api/v2/orders).
• Schema-driven I/O — Marshmallow or Pydantic validating every inbound payload and serializing every response.
• Consistent error envelope — all error responses follow RFC 7807 "problem details" or a house-style JSON shape with code, message, details.
• Infrastructure concerns bolted on — Flask-CORS, Flask-Limiter with Redis, Flask-JWT-Extended, Flask-Smorest for auto-generated OpenAPI.
• Served via Gunicorn (sync workers behind a thread pool) or Gunicorn + gthread/uvicorn workers, fronted by nginx or an ALB.

This page is a production-oriented reference: every pattern below is something you should ship, not just something that compiles.

2. Routing & HTTP Methods

Routes are declared with @app.route or the resource-specific shortcuts (@bp.get, @bp.post). Prefer the shortcuts — they are clearer and avoid accidentally accepting the wrong verb.

from flask import Flask, request

app = Flask(__name__)

# URL converters: int, string (default), float, path, uuid
@app.get("/api/v1/users/<int:user_id>")
def get_user(user_id: int):
    return {"id": user_id}

@app.get("/api/v1/orders/<uuid:order_id>")
def get_order(order_id):
    # order_id is a uuid.UUID instance
    return {"id": str(order_id)}

@app.get("/api/v1/files/<path:filepath>")
def get_file(filepath: str):
    # path converter matches slashes; use sparingly
    return {"path": filepath}

# Explicit methods list — prefer the verb-specific decorators instead
@app.route("/api/v1/users", methods=["GET", "POST"])
def users_collection():
    if request.method == "POST":
        return {"created": True}, 201
    return {"users": []}

Trailing slash behavior. Flask treats /users/ and /users as different routes. If the route is registered with a trailing slash, a request without one gets a 308 redirect. If registered without, a request with a slash returns 404. Pick a convention (trailing slashes for collections is common) and stick to it. Disable the redirect with app.url_map.strict_slashes = False if you want both to work.

3. Request Handling

The thread-local flask.request object exposes everything about the inbound HTTP request. Use the right accessor for the content type:

from flask import request, abort

@app.post("/api/v1/echo")
def echo():
    # JSON body (Content-Type: application/json)
    if not request.is_json:
        abort(415, "Expected application/json")
    payload = request.get_json(silent=False)  # raises 400 on malformed JSON

    # Query string: /echo?limit=20&cursor=abc
    limit = request.args.get("limit", default=20, type=int)
    cursor = request.args.get("cursor")

    # Form data (application/x-www-form-urlencoded or multipart/form-data)
    name = request.form.get("name")

    # Uploaded files (multipart/form-data)
    upload = request.files.get("attachment")
    if upload:
        upload.save(f"/tmp/{upload.filename}")

    # Headers
    request_id = request.headers.get("X-Request-ID")
    auth = request.headers.get("Authorization", "")

    # Raw body (bytes) — useful for webhook signature verification
    raw = request.get_data(cache=True)

    return {"received": payload, "limit": limit, "cursor": cursor,
            "request_id": request_id, "raw_size": len(raw)}

Content negotiation. Use request.accept_mimetypes to honor the client's Accept header when a route can serve multiple formats:

@app.get("/api/v1/report/<int:report_id>")
def get_report(report_id: int):
    best = request.accept_mimetypes.best_match(
        ["application/json", "text/csv"]
    )
    if best == "text/csv":
        return Response(render_csv(report_id), mimetype="text/csv")
    return {"id": report_id, "rows": load_report(report_id)}

4. Response Construction

Flask view functions can return:

• A dict or list — auto-serialized to JSON with status 200.
• A tuple (body, status) or (body, status, headers).
• A Response object for full control.
• A string (treated as HTML) — almost never what you want in a REST API.

from flask import jsonify, make_response, Response, abort

@app.post("/api/v1/users")
def create_user():
    # Tuple form — most idiomatic for REST
    return {"id": 42, "email": "kevin@example.com"}, 201, {
        "Location": "/api/v1/users/42",
        "X-Request-ID": request.headers.get("X-Request-ID", ""),
    }

@app.get("/api/v1/users/<int:uid>")
def get_user(uid: int):
    user = db.find_user(uid)
    if user is None:
        abort(404, description="User not found")
    # jsonify is equivalent to returning a dict in modern Flask, but explicit
    resp = jsonify(user.to_dict())
    resp.headers["Cache-Control"] = "private, max-age=60"
    resp.headers["ETag"] = user.etag()
    return resp

@app.delete("/api/v1/users/<int:uid>")
def delete_user(uid: int):
    db.delete_user(uid)
    return "", 204  # No Content — body must be empty

@app.get("/api/v1/exports/<int:job_id>.csv")
def download_csv(job_id: int):
    csv_bytes = build_csv(job_id)
    resp = make_response(csv_bytes)
    resp.headers["Content-Type"] = "text/csv"
    resp.headers["Content-Disposition"] = f'attachment; filename="export-{job_id}.csv"'
    return resp

flask.abort(status, description=...) raises an HTTPException that can be caught by a registered errorhandler. Prefer raising custom exceptions (see section 7) so callers get a consistent error envelope.

5. Blueprints & Project Structure

Blueprints are how Flask scales beyond a single-file script. Group routes by resource, register them inside the application factory, and mount them under a versioned URL prefix.

myapi/
  app/
    __init__.py           # create_app() factory
    extensions.py         # db, migrate, jwt, limiter, cors instances
    blueprints/
      __init__.py
      users.py
      auth.py
      orders.py
    models.py             # SQLAlchemy models
    schemas.py            # Marshmallow / Pydantic schemas
    errors.py             # Custom exception classes + handlers
    config.py             # DevConfig, ProdConfig, TestConfig
  tests/
    conftest.py
    test_users.py
  wsgi.py                 # Gunicorn entrypoint
  pyproject.toml
  Dockerfile

Application factory — never import a module-level app when you want tests to spin up isolated instances.

# app/__init__.py
from flask import Flask
from .extensions import db, migrate, jwt, limiter, cors
from .blueprints.users import users_bp
from .blueprints.auth import auth_bp
from .blueprints.orders import orders_bp
from .errors import register_error_handlers

def create_app(config_object: str = "app.config.ProdConfig") -> Flask:
    app = Flask(__name__)
    app.config.from_object(config_object)

    # Bind extensions
    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)
    limiter.init_app(app)
    cors.init_app(app, resources={r"/api/*": {"origins": app.config["CORS_ORIGINS"]}})

    # Register blueprints — versioned prefix per API generation
    app.register_blueprint(users_bp,  url_prefix="/api/v1/users")
    app.register_blueprint(auth_bp,   url_prefix="/api/v1/auth")
    app.register_blueprint(orders_bp, url_prefix="/api/v2/orders")  # v2 rollout

    register_error_handlers(app)
    return app

# app/blueprints/users.py
from flask import Blueprint, request
from ..schemas import UserCreateSchema, UserSchema
from ..models import User
from ..extensions import db

users_bp = Blueprint("users", __name__)
_in  = UserCreateSchema()
_out = UserSchema()

@users_bp.get("/<int:user_id>")
def get_user(user_id: int):
    user = db.session.get(User, user_id) or abort(404)
    return _out.dump(user)

@users_bp.post("")
def create_user():
    data = _in.load(request.get_json(force=True))  # raises ValidationError
    user = User(**data)
    db.session.add(user)
    db.session.commit()
    return _out.dump(user), 201, {"Location": f"/api/v1/users/{user.id}"}

6. Input Validation

Never trust request.json. Every field that reaches the database must pass through a schema. Two ecosystems dominate:

Marshmallow — mature, declarative, pairs well with Flask-Smorest for OpenAPI generation.

# app/schemas.py
from marshmallow import Schema, fields, validate, ValidationError, post_load

class UserCreateSchema(Schema):
    email = fields.Email(required=True)
    name  = fields.String(required=True, validate=validate.Length(min=1, max=120))
    age   = fields.Integer(validate=validate.Range(min=13, max=130))
    role  = fields.String(validate=validate.OneOf(["admin", "member", "guest"]),
                          load_default="member")

class UserSchema(Schema):
    id    = fields.Integer(dump_only=True)
    email = fields.Email()
    name  = fields.String()
    role  = fields.String()
    created_at = fields.DateTime(dump_only=True)

Pydantic v2 — faster, better type inference, widely used elsewhere. Pair with flask-pydantic or a small custom decorator:

from functools import wraps
from pydantic import BaseModel, EmailStr, Field, ValidationError
from flask import request, jsonify

class UserCreate(BaseModel):
    email: EmailStr
    name:  str = Field(min_length=1, max_length=120)
    age:   int | None = Field(default=None, ge=13, le=130)
    role:  str = Field(default="member", pattern="^(admin|member|guest)$")

def validate_body(model):
    def deco(view):
        @wraps(view)
        def inner(*args, **kwargs):
            try:
                payload = model.model_validate(request.get_json(force=True))
            except ValidationError as e:
                return jsonify({"code": "validation_error",
                                "errors": e.errors()}), 422
            return view(payload, *args, **kwargs)
        return inner
    return deco

@users_bp.post("")
@validate_body(UserCreate)
def create_user(payload: UserCreate):
    user = User(**payload.model_dump())
    db.session.add(user); db.session.commit()
    return user.to_dict(), 201

7. Error Handling

Every API should emit errors in a single, predictable shape. RFC 7807 ("Problem Details for HTTP APIs") is the closest thing to a standard: type, title, status, detail, instance.

# app/errors.py
from flask import jsonify
from werkzeug.exceptions import HTTPException
from marshmallow import ValidationError

class APIError(Exception):
    status_code = 400
    code = "bad_request"
    def __init__(self, message: str, *, status_code: int | None = None,
                 code: str | None = None, details: dict | None = None):
        super().__init__(message)
        self.message = message
        self.status_code = status_code or self.status_code
        self.code = code or self.code
        self.details = details or {}

class NotFound(APIError):    status_code = 404; code = "not_found"
class Conflict(APIError):    status_code = 409; code = "conflict"
class Unauthorized(APIError): status_code = 401; code = "unauthorized"
class Forbidden(APIError):   status_code = 403; code = "forbidden"

def _problem(status: int, title: str, code: str, detail: str = "", **extra):
    body = {
        "type":   f"https://errors.example.com/{code}",
        "title":  title,
        "status": status,
        "code":   code,
        "detail": detail,
        **extra,
    }
    return jsonify(body), status, {"Content-Type": "application/problem+json"}

def register_error_handlers(app):
    @app.errorhandler(APIError)
    def _api_error(e: APIError):
        return _problem(e.status_code, e.code.replace("_", " ").title(),
                        e.code, e.message, details=e.details)

    @app.errorhandler(ValidationError)
    def _marshmallow_error(e: ValidationError):
        return _problem(422, "Unprocessable Entity", "validation_error",
                        "Schema validation failed", errors=e.messages)

    @app.errorhandler(HTTPException)
    def _http_error(e: HTTPException):
        return _problem(e.code, e.name, e.name.lower().replace(" ", "_"),
                        e.description or "")

    @app.errorhandler(Exception)
    def _unhandled(e: Exception):
        app.logger.exception("unhandled")
        return _problem(500, "Internal Server Error", "internal_error",
                        "An unexpected error occurred")

8. HTTP Status Codes

Consistent status codes are part of the API contract. Teams invent their own meanings at their peril.

9. Authentication

Three common patterns: API keys (service-to-service), Bearer/JWT (SPAs and mobile apps), and session cookies (same-origin web apps). For a REST API the first two dominate.

# app/blueprints/auth.py
from flask import Blueprint, request, current_app, abort
from flask_jwt_extended import (
    JWTManager, create_access_token, create_refresh_token,
    jwt_required, get_jwt_identity, get_jwt,
)
from ..models import User
from ..extensions import jwt

auth_bp = Blueprint("auth", __name__)

@auth_bp.post("/login")
def login():
    body = request.get_json(force=True)
    user = User.query.filter_by(email=body["email"]).first()
    if user is None or not user.verify_password(body["password"]):
        abort(401, "Invalid credentials")
    claims = {"role": user.role, "tenant_id": user.tenant_id}
    return {
        "access_token":  create_access_token(identity=user.id, additional_claims=claims),
        "refresh_token": create_refresh_token(identity=user.id),
        "token_type":    "Bearer",
        "expires_in":    3600,
    }

@auth_bp.post("/refresh")
@jwt_required(refresh=True)
def refresh():
    uid = get_jwt_identity()
    return {"access_token": create_access_token(identity=uid)}

# Protected route with role-based authorization
from functools import wraps

def require_role(*roles):
    def deco(fn):
        @wraps(fn)
        @jwt_required()
        def inner(*args, **kwargs):
            claims = get_jwt()
            if claims.get("role") not in roles:
                abort(403, "Insufficient role")
            return fn(*args, **kwargs)
        return inner
    return deco

@users_bp.delete("/<int:user_id>")
@require_role("admin")
def delete_user(user_id: int):
    User.query.filter_by(id=user_id).delete()
    db.session.commit()
    return "", 204

For API keys, prefer a before_request hook on the blueprint that looks up the key in a Redis-backed cache keyed by a SHA-256 of the key string (never store raw keys).

10. CORS

CORS misconfiguration is one of the most common security bugs in REST APIs. Do not use origins="*" with supports_credentials=True — browsers will reject it anyway, and leaving it lax invites trouble.

# app/extensions.py
from flask_cors import CORS

cors = CORS()

# In create_app():
cors.init_app(
    app,
    resources={
        r"/api/*": {
            "origins": [
                "https://app.example.com",
                "https://admin.example.com",
            ],
            "methods": ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
            "allow_headers": ["Authorization", "Content-Type", "X-Request-ID"],
            "expose_headers": ["X-RateLimit-Remaining", "X-Request-ID", "Link"],
            "supports_credentials": True,
            "max_age": 600,  # seconds the browser may cache the preflight
        }
    },
)

11. Rate Limiting

Flask-Limiter with a Redis backend scales across multiple Gunicorn workers and hosts. Keys should usually be the authenticated user ID — fall back to IP for unauthenticated requests.

# app/extensions.py
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request

def rate_limit_key():
    try:
        verify_jwt_in_request(optional=True)
        uid = get_jwt_identity()
        if uid:
            return f"user:{uid}"
    except Exception:
        pass
    return f"ip:{get_remote_address()}"

limiter = Limiter(
    key_func=rate_limit_key,
    storage_uri="redis://redis.internal:6379/1",
    strategy="moving-window",
    default_limits=["1000/hour", "60/minute"],
    headers_enabled=True,  # X-RateLimit-* headers on every response
)

# Per-route override
@auth_bp.post("/login")
@limiter.limit("5 per minute; 20 per hour")
def login():
    ...

12. Pagination

Two patterns. Offset/limit is simple but breaks on large or mutating datasets — skipping 10,000 rows on every page is slow, and inserts shift pages. Cursor-based pagination is strongly preferred for any collection that can grow unboundedly.

import base64, json
from flask import url_for

def _encode_cursor(created_at, id_):
    return base64.urlsafe_b64encode(
        json.dumps([created_at.isoformat(), id_]).encode()
    ).decode()

def _decode_cursor(token: str):
    raw = json.loads(base64.urlsafe_b64decode(token.encode()))
    return raw[0], raw[1]

@users_bp.get("")
def list_users():
    limit  = min(request.args.get("limit", 50, type=int), 200)
    cursor = request.args.get("cursor")

    q = User.query.order_by(User.created_at.desc(), User.id.desc())
    if cursor:
        ts, last_id = _decode_cursor(cursor)
        q = q.filter(
            (User.created_at < ts) |
            ((User.created_at == ts) & (User.id < last_id))
        )
    rows = q.limit(limit + 1).all()

    has_more = len(rows) > limit
    page = rows[:limit]
    next_cursor = (_encode_cursor(page[-1].created_at, page[-1].id)
                   if has_more else None)

    headers = {}
    if next_cursor:
        next_url = url_for("users.list_users", limit=limit,
                           cursor=next_cursor, _external=True)
        headers["Link"] = f'<{next_url}>; rel="next"'

    return {"data": [u.to_dict() for u in page],
            "next_cursor": next_cursor}, 200, headers

13. OpenAPI / Swagger

Flask-Smorest generates OpenAPI 3.1 directly from Marshmallow schemas and blueprint decorators. It replaces Flask-RESTful and Flask-RESTX for most new projects.

from flask_smorest import Api, Blueprint
from marshmallow import Schema, fields

class UserArgs(Schema):
    limit = fields.Integer(load_default=50)
    cursor = fields.String()

class UserOut(Schema):
    id = fields.Integer()
    email = fields.Email()
    name = fields.String()

app.config["API_TITLE"] = "MyAPI"
app.config["API_VERSION"] = "v1"
app.config["OPENAPI_VERSION"] = "3.1.0"
app.config["OPENAPI_URL_PREFIX"] = "/docs"
app.config["OPENAPI_SWAGGER_UI_PATH"] = "/swagger"
app.config["OPENAPI_SWAGGER_UI_URL"] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"

api = Api(app)
users_bp = Blueprint("users", "users", url_prefix="/api/v1/users",
                     description="User management")

@users_bp.route("")
class Users(MethodView):
    @users_bp.arguments(UserArgs, location="query")
    @users_bp.response(200, UserOut(many=True))
    def get(self, args):
        return User.query.limit(args["limit"]).all()

    @users_bp.arguments(UserCreateSchema)
    @users_bp.response(201, UserOut)
    def post(self, data):
        user = User(**data); db.session.add(user); db.session.commit()
        return user

api.register_blueprint(users_bp)

The OpenAPI JSON is served at /docs/openapi.json and Swagger UI at /docs/swagger. Export the JSON into version control and diff it on every PR to catch accidental contract changes.

14. Testing REST Endpoints

app.test_client() gives you an in-process HTTP client — no sockets, no Gunicorn. Combined with pytest fixtures it is fast enough to run thousands of tests in under a minute.

# tests/conftest.py
import pytest
from app import create_app
from app.extensions import db as _db

@pytest.fixture(scope="session")
def app():
    app = create_app("app.config.TestConfig")
    with app.app_context():
        _db.create_all()
        yield app
        _db.drop_all()

@pytest.fixture()
def client(app):
    return app.test_client()

@pytest.fixture()
def auth_headers(client):
    resp = client.post("/api/v1/auth/login",
                       json={"email": "admin@example.com", "password": "secret"})
    token = resp.get_json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

# tests/test_users.py
def test_create_user_returns_201(client, auth_headers):
    resp = client.post("/api/v1/users",
                       json={"email": "new@example.com", "name": "New"},
                       headers=auth_headers)
    assert resp.status_code == 201
    assert resp.headers["Location"].startswith("/api/v1/users/")
    body = resp.get_json()
    assert body["email"] == "new@example.com"

def test_invalid_email_returns_422(client, auth_headers):
    resp = client.post("/api/v1/users",
                       json={"email": "nope", "name": "X"},
                       headers=auth_headers)
    assert resp.status_code == 422
    assert resp.get_json()["code"] == "validation_error"

def test_unauthenticated_returns_401(client):
    resp = client.get("/api/v1/users/1")
    assert resp.status_code == 401

def test_rate_limit(client, auth_headers):
    for _ in range(60):
        client.get("/api/v1/users", headers=auth_headers)
    resp = client.get("/api/v1/users", headers=auth_headers)
    assert resp.status_code == 429
    assert "Retry-After" in resp.headers

Run tests against a throwaway SQLite or a Docker-launched Postgres:

pytest -x -q --cov=app --cov-report=term-missing
# Or in CI, with Postgres:
docker run --rm -d -p 5432:5432 -e POSTGRES_PASSWORD=test --name pg postgres:16
DATABASE_URL=postgresql://postgres:test@localhost:5432/postgres pytest

A Flask REST API built along these lines — factory + blueprints + schema validation + consistent errors + JWT + rate limiting + cursor pagination + OpenAPI — is the baseline I ship for every production service. Everything else (caching, background jobs, webhooks, tracing) is layered on top of this skeleton.