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:
create_app()) returning a configured
Flask instance — required for testability and multi-config deployments./api/v1/users, /api/v2/orders).code, message,
details.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.
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.
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)}
Flask view functions can return:
dict or list — auto-serialized to JSON with status 200.(body, status) or (body, status, headers).Response object for full control.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.
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}"}
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
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")
Consistent status codes are part of the API contract. Teams invent their own meanings at their peril.
| Code | Name | When to use |
| 200 | OK | Successful GET, PUT, PATCH with a body |
| 201 | Created | Resource created; include Location header |
| 202 | Accepted | Async job queued; return job URL for polling |
| 204 | No Content | Successful DELETE or PUT with no response body |
| 400 | Bad Request | Malformed JSON, missing headers, unparseable input |
| 401 | Unauthorized | No or invalid credentials (really "unauthenticated") |
| 403 | Forbidden | Authenticated but lacks permission |
| 404 | Not Found | Resource does not exist (or hidden from this caller) |
| 409 | Conflict | Duplicate key, optimistic-lock failure, state conflict |
| 422 | Unprocessable Entity | Well-formed JSON that fails schema validation |
| 429 | Too Many Requests | Rate limit exceeded; include Retry-After |
| 500 | Internal Server Error | Unhandled server exception; never leak stack traces |
| 503 | Service Unavailable | Dependency down, maintenance, or overload |
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).
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
}
},
)
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():
...
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
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.
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.