Flask - Building REST APIs the Right Way

Reading time: ~35 minutes | Level: Intermediate → Engineering

Before reading further, predict what each of these three requests returns:

from flask import Flask, request

app = Flask(__name__)

@app.route("/divide")
def divide():
    a = int(request.args.get("a"))
    b = int(request.args.get("b"))
    return {"result": a / b}

# GET /divide?a=10&b=2   → ?
# GET /divide?a=10&b=0   → ?
# GET /divide?a=10       → ?
# GET /divide?a=abc&b=2  → ?

Show Answer

GET /divide?a=10&b=2   → {"result": 5.0}  - 200 OK  (works correctly)
GET /divide?a=10&b=0   → 500 Internal Server Error  (ZeroDivisionError: division by zero)
GET /divide?a=10       → 500 Internal Server Error  (TypeError: int() argument is None)
GET /divide?a=abc&b=2  → 500 Internal Server Error  (ValueError: invalid literal for int())

Three different logical failures - missing parameter, invalid type, divide-by-zero - all surface as 500 Internal Server Error with an HTML traceback in development mode and a bare 500 with no useful body in production mode.

A production API must handle all three explicitly:

• Missing parameter → 400 Bad Request with a message identifying the missing field
• Invalid type → 400 Bad Request with a message about the invalid value
• Divide by zero → 422 Unprocessable Entity (valid input, invalid operation)

This lesson shows how to build the error handling layer that prevents your Flask API from silently serving 500 for every unhandled edge case.

Flask is a micro-framework - it gives you routing, a request/response cycle, and a WSGI interface. Everything else (database, auth, validation, serialization) is your responsibility. This is its strength and its danger. Teams that understand Flask build clean, testable APIs with explicit control over every layer. Teams that do not understand it build app.py monoliths with global state, context errors, and test suites that cannot run in parallel.

What You Will Learn

• Flask's design philosophy: WSGI, micro-framework, extension model
• Application factory pattern: why create_app() is non-negotiable for testing
• Request context: how request, g, and session work as context-local proxies
• Routing: URL converters, HTTP methods, url_for, reverse URL generation
• Request and response objects: every useful attribute and method
• Error handlers: catching exceptions globally, returning consistent RFC 7807 error bodies
• Blueprints: splitting routes across modules without circular imports
• Testing Flask: app.test_client(), app.test_request_context()
• Configuration: environment-based config patterns
• Extension ecosystem: what to reach for and why
• When to choose Flask vs FastAPI

Prerequisites

• Python classes, decorators, and context managers (Modules 01–02)
• HTTP methods and status codes (Lesson 01 of this module)
• REST principles (Lesson 02 of this module)
• Basic familiarity with pytest

Part 1 - Flask's Design Philosophy

Flask is built on two libraries:

• Werkzeug: the WSGI utility library - HTTP parsing, routing, request/response objects, the WSGI server interface
• Jinja2: the template engine (not used in REST APIs, but part of Flask's heritage)

WSGI (Web Server Gateway Interface) is Python's standard interface between web servers (nginx, gunicorn, uWSGI) and Python web applications. A WSGI application is a callable that receives an environ dict (the HTTP request) and a start_response callable, and returns an iterable of response bytes.

Flask wraps this:

# What WSGI actually looks like (Flask hides this)
def wsgi_app(environ: dict, start_response) -> list[bytes]:
    # environ contains: REQUEST_METHOD, PATH_INFO, QUERY_STRING, headers, etc.
    start_response("200 OK", [("Content-Type", "application/json")])
    return [b'{"result": 42}']

# Flask gives you this instead:
@app.route("/result")
def get_result():
    return {"result": 42}

Flask's "micro" means: it does not include an ORM, does not mandate a project structure, does not include form validation or serialization. This differs from Django, which includes all of these. Flask is micro because it trusts you to choose the right tool for each layer.

Part 2 - Application Factory Pattern

The Wrong Way: Module-Level app

# app.py - the approach that breaks everything
from flask import Flask
from extensions import db

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///dev.db"
db.init_app(app)

@app.route("/users")
def list_users():
    return {"users": []}

# Problems:
# 1. Tests cannot create isolated instances - every test shares this one app
# 2. Cannot have different configs for test/dev/prod without environment hacks
# 3. Circular imports: if models.py imports from app.py and app.py imports from models.py
# 4. Extensions initialized at import time - no control over initialization order

The Right Way: Application Factory

# app/__init__.py - the factory function

from flask import Flask
from .extensions import db, migrate, jwt
from .config import Config


def create_app(config_object: object = None) -> Flask:
    """
    Application factory.

    Args:
        config_object: configuration object or class. Defaults to Config.

    Returns:
        Configured Flask application instance.
    """
    app = Flask(__name__)

    # Load configuration
    app.config.from_object(config_object or Config)

    # Initialize extensions (in order - db before migrate)
    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)

    # Register blueprints
    from .users.routes import users_bp
    from .orders.routes import orders_bp
    from .health import health_bp

    app.register_blueprint(users_bp, url_prefix="/v1/users")
    app.register_blueprint(orders_bp, url_prefix="/v1/orders")
    app.register_blueprint(health_bp)

    # Register error handlers
    from .errors import register_error_handlers
    register_error_handlers(app)

    return app

# run.py - entry point
from app import create_app

app = create_app()

if __name__ == "__main__":
    app.run(debug=True)

# tests/conftest.py - test setup
import pytest
from app import create_app
from app.config import TestConfig

@pytest.fixture(scope="session")
def app():
    app = create_app(TestConfig)
    return app

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

@pytest.fixture
def runner(app):
    return app.test_cli_runner()

:::tip Always Use the Application Factory Pattern The factory pattern (create_app()) makes every aspect of Flask testable: you create a fresh app instance per test session with a test database, isolated config, and no shared state. Without it, tests share the same app instance, the same database URL, and the same extension state - creating test interdependencies that cause flaky, order-dependent test failures. In new Flask projects, use the factory pattern from line one. :::

Part 3 - Request Context: How Proxies Work

Flask's request, g, and session are not regular module-level variables. They are context-local proxies - objects that look up the actual request or session data from a per-request context stack.

When gunicorn handles 20 concurrent requests, each request has its own RequestContext on the context stack. When view_function_1 reads request.args, Flask looks up the context for view_function_1's request - not any other concurrent request's context.

from flask import request, g, session

# These look like module-level variables but they are context proxies
# Accessing them outside a request context raises RuntimeError

@app.before_request
def load_user():
    """Runs before every request. Attach data to g for use in views."""
    token = request.headers.get("Authorization", "").removeprefix("Bearer ")
    if token:
        g.current_user = verify_jwt(token)  # g lives for this request only
    else:
        g.current_user = None

@app.route("/profile")
def profile():
    if not g.current_user:
        return {"error": "Not authenticated"}, 401
    return {"user": g.current_user.to_dict()}

:::warning request Is a Context-Local Proxy: Do Not Use It Outside Request Context

# WRONG - accessing request outside a request context
from flask import request

def get_current_user():
    # This function might be called from a background task or CLI command
    # where there is no active request context - raises RuntimeError
    return User.query.get(request.json["user_id"])

# RIGHT - pass the data explicitly, not via the proxy
def get_current_user(user_id: int):
    return User.query.get(user_id)

# In views:
@app.route("/user")
def user_view():
    user_id = request.json["user_id"]  # access proxy in view (correct context)
    user = get_current_user(user_id)   # pass data to pure function
    return jsonify(user.to_dict())

The rule: access request, g, and session only inside view functions, before_request hooks, and after_request hooks - where Flask guarantees an active request context. Pure functions and utility modules must receive data as arguments. :::

Part 4 - Routing

Basic Routes

from flask import Flask, request, jsonify, url_for

app = Flask(__name__)

# Simple route
@app.route("/health")
def health():
    return {"status": "ok"}

# URL converters - type-checking built into the route
@app.route("/users/<int:user_id>")         # int: casts to Python int
def get_user(user_id: int):
    return {"user_id": user_id}

@app.route("/files/<path:filepath>")       # path: matches slashes
def get_file(filepath: str):
    return {"path": filepath}

@app.route("/tokens/<uuid:token_id>")      # uuid: validates UUID format
def get_token(token_id):
    return {"token": str(token_id)}

# Multiple methods on one route
@app.route("/users", methods=["GET", "POST"])
def users():
    if request.method == "GET":
        return jsonify(list_users())
    if request.method == "POST":
        return jsonify(create_user(request.get_json())), 201

URL Converters Reference

Converter	Type	Example
string (default)	str, no slashes	<name> matches alice
int	int	<int:id> matches 123
float	float	<float:rate> matches 3.14
path	str, with slashes	<path:filepath> matches a/b/c
uuid	uuid.UUID	<uuid:id> validates UUID format

Reverse URL Generation with url_for

from flask import url_for

# Inside a view or test - generate URLs from function names, not strings
@app.route("/users/<int:user_id>")
def get_user(user_id: int):
    return {"user": {"id": user_id}}

@app.route("/users", methods=["POST"])
def create_user():
    user = User.create(request.get_json())
    # Generate the Location URL without hardcoding the path
    location = url_for("get_user", user_id=user.id, _external=True)
    return jsonify(user.to_dict()), 201, {"Location": location}

# url_for("get_user", user_id=123) → "/users/123"
# url_for("get_user", user_id=123, _external=True) → "https://api.example.com/users/123"

url_for references the function name, not the URL string. When you change /users/<int:user_id> to /v2/users/<int:user_id>, every url_for("get_user", ...) call updates automatically.

Part 5 - Request Object

from flask import request

@app.route("/users", methods=["GET"])
def list_users():
    # Query string parameters: /users?active=true&role=admin&limit=25
    active = request.args.get("active", type=str)      # "true" or None
    role = request.args.get("role", default="user")    # str with default
    limit = request.args.get("limit", default=25, type=int)  # cast to int

    # All query params
    all_params = dict(request.args)    # {"active": "true", "role": "admin", "limit": "25"}

    return jsonify(get_users(active=active, role=role, limit=limit))


@app.route("/users", methods=["POST"])
def create_user():
    # JSON body - preferred for REST APIs
    data = request.get_json()          # parses body as JSON; returns None if not JSON
    data = request.get_json(silent=True)     # returns None instead of 400 on parse error
    data = request.get_json(force=True)      # parses even if Content-Type is not application/json

    # Form data - for HTML form submissions
    name = request.form.get("name")

    # Raw body
    raw = request.data                  # bytes
    raw_str = request.data.decode()     # string

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

    # Method, URL, path
    method = request.method             # "POST"
    url = request.url                   # "https://api.example.com/users"
    path = request.path                 # "/users"
    remote_addr = request.remote_addr   # client IP

    # File uploads
    uploaded_file = request.files.get("avatar")
    if uploaded_file:
        uploaded_file.save(f"/uploads/{uploaded_file.filename}")

    return jsonify({"created": True}), 201

Part 6 - Response Patterns

from flask import jsonify, make_response, abort, Response

# 1. Return a dict (Flask 2.2+ auto-jsonifies dicts)
@app.route("/users/<int:user_id>")
def get_user(user_id):
    return {"id": user_id, "name": "Alice"}   # 200, Content-Type: application/json

# 2. jsonify - explicit, works in all Flask versions
@app.route("/users/<int:user_id>")
def get_user_v2(user_id):
    return jsonify({"id": user_id, "name": "Alice"})

# 3. Return tuple (body, status_code)
@app.route("/users", methods=["POST"])
def create_user():
    user = create(request.get_json())
    return jsonify(user), 201

# 4. Return tuple (body, status_code, headers)
@app.route("/users", methods=["POST"])
def create_user_with_headers():
    user = create(request.get_json())
    return jsonify(user), 201, {"Location": f"/users/{user['id']}"}

# 5. make_response - full control
@app.route("/users/<int:user_id>")
def get_user_custom(user_id):
    response = make_response(jsonify({"id": user_id}))
    response.status_code = 200
    response.headers["Cache-Control"] = "max-age=60, private"
    response.headers["ETag"] = f'"{compute_etag(user_id)}"'
    return response

# 6. abort - immediately raise an HTTP error (intercepted by error handlers)
@app.route("/admin/users")
def admin_list_users():
    if not is_admin(g.current_user):
        abort(403)    # raises Forbidden, caught by @app.errorhandler(403)
    return jsonify(list_all_users())

:::danger return {"result": value} Auto-jsonifies Only in Flask 2.2+ In Flask versions before 2.2, returning a dict from a view function does not work as expected - it may return a string representation rather than a JSON response, or raise a TypeError. If your team's Flask version is not pinned or you are writing a library, always use jsonify() explicitly. It is defensive and version-agnostic. :::

Part 7 - Error Handlers

The Problem: Unhandled Exceptions Become Bare 500s

# Without error handlers, this is what happens in production:
# - ZeroDivisionError → 500 with HTML traceback (or bare 500 in production mode)
# - KeyError → 500
# - SQLAlchemy.exc.NoResultFound → 500
# - requests.exceptions.Timeout → 500
#
# All of these expose implementation details to clients and
# return non-JSON bodies that JSON clients cannot parse.

Defining Global Error Handlers

# errors.py

from flask import Flask, jsonify
from werkzeug.exceptions import HTTPException


def register_error_handlers(app: Flask) -> None:

    @app.errorhandler(400)
    def bad_request(error):
        return jsonify({
            "type": "https://api.example.com/errors/bad-request",
            "title": "Bad Request",
            "status": 400,
            "detail": str(error.description),
        }), 400, {"Content-Type": "application/problem+json"}

    @app.errorhandler(401)
    def unauthorized(error):
        return jsonify({
            "type": "https://api.example.com/errors/unauthorized",
            "title": "Authentication Required",
            "status": 401,
            "detail": "Valid authentication credentials are required.",
        }), 401, {"Content-Type": "application/problem+json"}

    @app.errorhandler(403)
    def forbidden(error):
        return jsonify({
            "type": "https://api.example.com/errors/forbidden",
            "title": "Forbidden",
            "status": 403,
            "detail": "You do not have permission to access this resource.",
        }), 403, {"Content-Type": "application/problem+json"}

    @app.errorhandler(404)
    def not_found(error):
        return jsonify({
            "type": "https://api.example.com/errors/not-found",
            "title": "Not Found",
            "status": 404,
            "detail": str(error.description),
        }), 404, {"Content-Type": "application/problem+json"}

    @app.errorhandler(405)
    def method_not_allowed(error):
        return jsonify({
            "type": "https://api.example.com/errors/method-not-allowed",
            "title": "Method Not Allowed",
            "status": 405,
        }), 405, {"Content-Type": "application/problem+json"}

    @app.errorhandler(422)
    def unprocessable_entity(error):
        return jsonify({
            "type": "https://api.example.com/errors/validation-error",
            "title": "Validation Error",
            "status": 422,
            "detail": str(error.description),
        }), 422, {"Content-Type": "application/problem+json"}

    @app.errorhandler(429)
    def rate_limited(error):
        return jsonify({
            "type": "https://api.example.com/errors/rate-limited",
            "title": "Too Many Requests",
            "status": 429,
        }), 429, {
            "Content-Type": "application/problem+json",
            "Retry-After": "60",
        }

    @app.errorhandler(HTTPException)
    def handle_http_exception(error: HTTPException):
        """Catch-all for werkzeug HTTP exceptions not handled above."""
        return jsonify({
            "type": f"https://api.example.com/errors/{error.name.lower().replace(' ', '-')}",
            "title": error.name,
            "status": error.code,
            "detail": error.description,
        }), error.code, {"Content-Type": "application/problem+json"}

    @app.errorhandler(Exception)
    def handle_unexpected_exception(error: Exception):
        """Catch unhandled exceptions - log them, return generic 500."""
        import logging
        import traceback
        logging.getLogger(__name__).error(
            "Unhandled exception: %s\n%s",
            str(error),
            traceback.format_exc(),
        )
        return jsonify({
            "type": "https://api.example.com/errors/internal-server-error",
            "title": "Internal Server Error",
            "status": 500,
            "detail": "An unexpected error occurred. The error has been logged.",
        }), 500, {"Content-Type": "application/problem+json"}

Custom Application Exceptions

# exceptions.py - domain-specific exceptions that map to HTTP responses

class APIError(Exception):
    """Base class for all API errors. Carries HTTP status and RFC 7807 body."""
    status_code = 500
    error_type = "https://api.example.com/errors/internal-error"
    title = "Internal Server Error"

    def __init__(self, detail: str = None, **extensions):
        self.detail = detail
        self.extensions = extensions
        super().__init__(detail or self.title)

    def to_dict(self) -> dict:
        body = {
            "type": self.error_type,
            "title": self.title,
            "status": self.status_code,
        }
        if self.detail:
            body["detail"] = self.detail
        body.update(self.extensions)
        return body


class NotFoundError(APIError):
    status_code = 404
    error_type = "https://api.example.com/errors/not-found"
    title = "Not Found"


class ValidationError(APIError):
    status_code = 422
    error_type = "https://api.example.com/errors/validation-error"
    title = "Validation Error"


class ConflictError(APIError):
    status_code = 409
    error_type = "https://api.example.com/errors/conflict"
    title = "Conflict"


# Register the handler in create_app
def register_error_handlers(app: Flask) -> None:
    # ... existing handlers ...

    @app.errorhandler(APIError)
    def handle_api_error(error: APIError):
        return jsonify(error.to_dict()), error.status_code, {
            "Content-Type": "application/problem+json"
        }


# Usage in views - raise instead of returning error tuples
@app.route("/users/<int:user_id>")
def get_user(user_id):
    user = User.query.get(user_id)
    if not user:
        raise NotFoundError(f"No user exists with ID {user_id}")
    return jsonify(user.to_dict())

Part 8 - Blueprints: Modular Route Organisation

Blueprints split routes across modules without circular imports. Each blueprint is a collection of routes, error handlers, and template folders that can be registered on the app:

# users/routes.py

from flask import Blueprint, request, jsonify, g
from .models import User
from .schema import UserCreateSchema, UserUpdateSchema
from ..exceptions import NotFoundError, ValidationError, ConflictError
from ..auth import require_auth

users_bp = Blueprint("users", __name__)

@users_bp.route("/", methods=["GET"])
@require_auth
def list_users():
    """
    List users with optional filtering and cursor-based pagination.

    Query params:
        active (bool): filter by active status
        cursor (str): pagination cursor from previous response
        limit (int): items per page (default 25, max 100)
    """
    active_str = request.args.get("active")
    active = active_str.lower() == "true" if active_str else None
    cursor = request.args.get("cursor")
    limit = min(request.args.get("limit", default=25, type=int), 100)

    result = User.list_paginated(active=active, cursor=cursor, limit=limit)
    return jsonify(result)


@users_bp.route("/", methods=["POST"])
def create_user():
    data = request.get_json(silent=True)
    if not data:
        raise ValidationError("Request body must be valid JSON")

    errors = UserCreateSchema().validate(data)
    if errors:
        raise ValidationError("Invalid request data", fields=errors)

    if User.query.filter_by(email=data["email"]).first():
        raise ConflictError(f"A user with email '{data['email']}' already exists")

    user = User.create(data)
    return jsonify(user.to_dict()), 201, {
        "Location": f"/v1/users/{user.id}"
    }


@users_bp.route("/<int:user_id>", methods=["GET"])
def get_user(user_id):
    user = User.query.get(user_id)
    if not user:
        raise NotFoundError(f"No user exists with ID {user_id}")
    return jsonify(user.to_dict())


@users_bp.route("/<int:user_id>", methods=["PATCH"])
@require_auth
def update_user(user_id):
    user = User.query.get(user_id)
    if not user:
        raise NotFoundError(f"No user exists with ID {user_id}")

    data = request.get_json(silent=True)
    if not data:
        raise ValidationError("Request body must be valid JSON")

    errors = UserUpdateSchema().validate(data)
    if errors:
        raise ValidationError("Invalid update data", fields=errors)

    user.update(data)
    return jsonify(user.to_dict())


@users_bp.route("/<int:user_id>", methods=["DELETE"])
@require_auth
def delete_user(user_id):
    user = User.query.get(user_id)
    if not user:
        raise NotFoundError(f"No user exists with ID {user_id}")
    user.delete()
    return "", 204

Part 9 - Configuration

# config.py

import os


class Config:
    """Base configuration. Values shared across all environments."""
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-CHANGE-IN-PRODUCTION")
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JSON_SORT_KEYS = False
    MAX_CONTENT_LENGTH = 16 * 1024 * 1024   # 16MB max upload


class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL", "postgresql://localhost/myapp_dev"
    )
    SQLALCHEMY_ECHO = True   # log all SQL in development


class ProductionConfig(Config):
    DEBUG = False
    TESTING = False
    SQLALCHEMY_DATABASE_URI = os.environ["DATABASE_URL"]  # must be set; no default
    SQLALCHEMY_POOL_SIZE = 10
    SQLALCHEMY_MAX_OVERFLOW = 5


class TestConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"   # in-memory for speed
    SECRET_KEY = "test-secret-key"
    WTF_CSRF_ENABLED = False


# Environment-based selection
config_by_name = {
    "development": DevelopmentConfig,
    "production": ProductionConfig,
    "test": TestConfig,
}

def get_config():
    env = os.environ.get("FLASK_ENV", "development")
    return config_by_name.get(env, DevelopmentConfig)

# create_app uses the config
def create_app(config_object=None):
    app = Flask(__name__)
    app.config.from_object(config_object or get_config())
    ...

Part 10 - Testing Flask Applications

# tests/test_users.py

import pytest
import json
from app import create_app
from app.config import TestConfig
from app.extensions import db as _db
from app.models import User


@pytest.fixture(scope="session")
def app():
    """Create application for the whole test session."""
    app = create_app(TestConfig)
    with app.app_context():
        _db.create_all()
        yield app
        _db.drop_all()


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


@pytest.fixture(autouse=True)
def clean_db(app):
    """Roll back all DB changes after each test."""
    with app.app_context():
        yield
        _db.session.rollback()
        for table in reversed(_db.metadata.sorted_tables):
            _db.session.execute(table.delete())
        _db.session.commit()


class TestGetUser:

    def test_get_existing_user(self, client, app):
        with app.app_context():
            user = User.create({"name": "Alice", "email": "[email protected]"})
            user_id = user.id

        response = client.get(f"/v1/users/{user_id}")

        assert response.status_code == 200
        data = response.get_json()
        assert data["id"] == user_id
        assert data["name"] == "Alice"
        assert data["email"] == "[email protected]"

    def test_get_nonexistent_user_returns_404(self, client):
        response = client.get("/v1/users/999999")

        assert response.status_code == 404
        data = response.get_json()
        # Verify RFC 7807 shape
        assert data["status"] == 404
        assert "type" in data
        assert "title" in data

    def test_content_type_is_json(self, client, app):
        with app.app_context():
            user = User.create({"name": "Bob", "email": "[email protected]"})
            user_id = user.id

        response = client.get(f"/v1/users/{user_id}")
        assert "application/json" in response.content_type


class TestCreateUser:

    def test_create_user_success(self, client):
        response = client.post(
            "/v1/users/",
            data=json.dumps({"name": "Charlie", "email": "[email protected]"}),
            content_type="application/json",
        )

        assert response.status_code == 201
        assert "Location" in response.headers
        data = response.get_json()
        assert data["name"] == "Charlie"
        assert "id" in data

    def test_create_duplicate_email_returns_409(self, client, app):
        with app.app_context():
            User.create({"name": "Diana", "email": "[email protected]"})

        response = client.post(
            "/v1/users/",
            data=json.dumps({"name": "Diana2", "email": "[email protected]"}),
            content_type="application/json",
        )

        assert response.status_code == 409
        data = response.get_json()
        assert data["status"] == 409

    def test_create_with_missing_email_returns_422(self, client):
        response = client.post(
            "/v1/users/",
            data=json.dumps({"name": "NoEmail"}),
            content_type="application/json",
        )

        assert response.status_code == 422

    def test_create_with_invalid_json_returns_422(self, client):
        response = client.post(
            "/v1/users/",
            data="not json at all",
            content_type="application/json",
        )

        assert response.status_code == 422

    def test_delete_returns_204_with_no_body(self, client, app):
        with app.app_context():
            user = User.create({"name": "Eve", "email": "[email protected]"})
            user_id = user.id

        response = client.delete(f"/v1/users/{user_id}")

        assert response.status_code == 204
        assert response.data == b""

Part 11 - Extension Ecosystem

Flask's power comes from its extension ecosystem. Key extensions for production APIs:

Extension	Purpose	Install
Flask-SQLAlchemy	SQLAlchemy ORM integration	pip install flask-sqlalchemy
Flask-Migrate	Alembic DB migrations via Flask CLI	pip install flask-migrate
Flask-JWT-Extended	JWT authentication, refresh tokens	pip install flask-jwt-extended
Flask-CORS	Cross-Origin Resource Sharing headers	pip install flask-cors
flask-smorest	Blueprint + OpenAPI spec generation	pip install flask-smorest
Flask-Limiter	Rate limiting with Redis backend	pip install flask-limiter
Flask-Caching	Response caching with Redis/Memcached	pip install flask-caching

# extensions.py - instantiate extensions here, init in create_app

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
from flask_cors import CORS
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()
cors = CORS()
limiter = Limiter(key_func=get_remote_address, default_limits=["200/hour"])

# In create_app:
# db.init_app(app)
# migrate.init_app(app, db)
# jwt.init_app(app)
# cors.init_app(app, resources={r"/v1/*": {"origins": ["https://app.example.com"]}})
# limiter.init_app(app)

:::note Flask Is Synchronous by Default Flask runs synchronous view functions in threads (via gunicorn worker threads). In Flask 2.0+, you can declare async def view functions, but Flask uses a thread executor - it does not run a true asyncio event loop. For genuinely async workloads (many concurrent connections, WebSockets, long-polling), FastAPI with uvicorn is the correct tool. Use Flask for synchronous APIs; use FastAPI for async-native workloads. :::

Part 12 - When to Choose Flask

Choose Flask when...	Consider FastAPI instead when...
Team has Flask expertise	Greenfield project with no existing codebase
Integrating with Flask-SQLAlchemy ecosystem	Type-safe request/response validation is required
Need full control over every layer	Auto-generated OpenAPI spec is a hard requirement
Simple CRUD API with modest traffic	High-concurrency or async I/O workloads
Teaching or prototyping	Client SDKs will be auto-generated from the spec
Existing Django project already on WSGI	WebSockets or server-sent events

Flask is not "lesser than" FastAPI - it is a different tool with different strengths. Flask's explicit, minimal design makes it excellent for teams that want full control and are comfortable assembling the right combination of extensions. FastAPI's type-annotation-driven design makes it excellent for teams that want automatic validation, serialization, and documentation with less boilerplate.

Graded Practice Challenges

Level 1 - Predict and Identify

Question 1: This Flask code will raise a RuntimeError in one specific situation. What is it?

from flask import Flask, request

app = Flask(__name__)

# Called from a Celery task (background, outside request context)
def send_welcome_email():
    user_id = request.json.get("user_id")   # line A
    send_email(user_id)

Show Answer

Line A raises RuntimeError: Working outside of request context when send_welcome_email() is called from a Celery task (or any other context where Flask has not set up a request context).

request is a context-local proxy. When accessed, it looks up the current request context on the context stack. Outside a Flask request handler, there is no request context - the proxy raises RuntimeError.

Fix: pass data as arguments, not via the proxy:

def send_welcome_email(user_id: int) -> None:
    """Pure function - takes explicit arguments, never touches Flask proxies."""
    send_email(user_id)

# In the view:
@app.route("/users", methods=["POST"])
def create_user():
    user_id = request.json.get("user_id")   # access proxy here (correct context)
    celery_task.delay(user_id)               # pass ID, not the request
    return jsonify({"queued": True}), 202

Question 2: What HTTP response does Flask send for a route registered with methods=["POST"] when a client sends a GET request to it?

Show Answer

Flask automatically returns 405 Method Not Allowed with an Allow: POST header in the response. This is Werkzeug's built-in behaviour - it detects that the URL exists but the method is not registered, and returns the correct HTTP error code without any view function being called.

With the error handler registered:

@app.errorhandler(405)
def method_not_allowed(error):
    return jsonify({
        "type": "https://api.example.com/errors/method-not-allowed",
        "title": "Method Not Allowed",
        "status": 405,
    }), 405

the response will be a JSON body. Without the handler, it will be an HTML response.

Question 3: What is the difference between these two Blueprint registrations?

# Option A
app.register_blueprint(users_bp, url_prefix="/v1/users")

# Option B
users_bp = Blueprint("users", __name__, url_prefix="/v1/users")
app.register_blueprint(users_bp)

Show Answer

Functionally, both produce the same URL prefix. The difference is where the url_prefix is defined:

• Option A: the prefix is defined at registration time in create_app(). The Blueprint itself is reusable - you could register the same blueprint at /v1/users in production and /test/users in testing.

• Option B: the prefix is baked into the Blueprint definition. Changing it requires editing the blueprint file.

Option A is preferred for the application factory pattern because it keeps deployment decisions (URL prefixes) in create_app() where all deployment configuration lives, not scattered through blueprint definitions.

Question 4: Why does this test fail even though the endpoint works correctly in manual testing?

from app import create_app

app = create_app()

def test_create_user():
    response = app.test_client().post("/v1/users/", json={"name": "Alice", "email": "[email protected]"})
    assert response.status_code == 201

Show Answer

The test fails because create_app() uses the default (development) config, which points to a real PostgreSQL database. The test:

1. Uses TestConfig if it exists - but this test does not pass TestConfig to create_app()
2. Uses the development database, which may not be running in CI
3. Creates real database records that persist between test runs

Fix - pass TestConfig explicitly:

import pytest
from app import create_app
from app.config import TestConfig
from app.extensions import db

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

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

def test_create_user(client):
    response = client.post("/v1/users/", json={"name": "Alice", "email": "[email protected]"})
    assert response.status_code == 201

This is exactly why the application factory pattern is non-negotiable for testing - it enables each test session to create a fresh app with the correct configuration.

Level 2 - Debug and Fix

Find and fix all problems in this Flask application:

# app.py
from flask import Flask, request, jsonify
import sqlite3

# Problem set 1: global app, direct DB connection
app = Flask(__name__)
conn = sqlite3.connect("users.db")

@app.route("/users")
def get_users():
    # Problem set 2: no error handling
    cursor = conn.cursor()
    users = cursor.execute("SELECT * FROM users").fetchall()
    return jsonify(users)   # returning raw tuples

@app.route("/users/<id>", methods=["DELETE"])
def delete_user(id):
    # Problem set 3: wrong method semantics, no status code handling
    cursor = conn.cursor()
    cursor.execute("DELETE FROM users WHERE id = ?", (id,))
    conn.commit()
    return jsonify({"success": True, "deleted": id}), 200

@app.route("/users", methods=["POST"])
def create_user():
    # Problem set 4: no validation, wrong status code
    data = request.json
    cursor = conn.cursor()
    cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)",
                   (data["name"], data["email"]))
    conn.commit()
    return jsonify({"created": True}), 200

if __name__ == "__main__":
    app.run(debug=True)

Show Solution

Problem 1 - Global app and persistent sqlite3.connect(): sqlite3.connect() at module level creates one connection that is shared by all threads. SQLite connections are not thread-safe in this mode. Use the application factory and Flask-SQLAlchemy for connection pooling.

Problem 2 - Raw tuples returned as JSON, no error handling: cursor.fetchall() returns a list of tuples. jsonify([(1, "Alice", "[email protected]")]) produces [[1, "Alice", "[email protected]"]] - no field names. Clients cannot use this.

Problem 3 - DELETE returns a body with status 200: DELETE should return 204 No Content with no body. Also no 404 handling.

Problem 4 - POST returns 200 instead of 201, no validation: Creating a resource should return 201 Created with a Location header. request.json can be None (missing or malformed body) - accessing data["name"] raises TypeError or KeyError → 500.

Fixed version:

# app/__init__.py - application factory
from flask import Flask
from .extensions import db
from .config import Config

def create_app(config_object=None):
    app = Flask(__name__)
    app.config.from_object(config_object or Config)
    db.init_app(app)

    from .users.routes import users_bp
    app.register_blueprint(users_bp, url_prefix="/v1/users")

    from .errors import register_error_handlers
    register_error_handlers(app)

    return app

# users/routes.py
from flask import Blueprint, request, jsonify
from ..extensions import db
from ..models import User
from ..exceptions import NotFoundError, ValidationError

users_bp = Blueprint("users", __name__)

@users_bp.route("/")
def get_users():
    users = User.query.all()
    return jsonify([u.to_dict() for u in users])   # dicts, not tuples

@users_bp.route("/<int:user_id>", methods=["DELETE"])
def delete_user(user_id):
    user = User.query.get(user_id)
    if not user:
        raise NotFoundError(f"User {user_id} not found")
    db.session.delete(user)
    db.session.commit()
    return "", 204   # No Content, no body

@users_bp.route("/", methods=["POST"])
def create_user():
    data = request.get_json(silent=True)
    if not data:
        raise ValidationError("Request body must be valid JSON")
    if not data.get("name"):
        raise ValidationError("Field 'name' is required")
    if not data.get("email"):
        raise ValidationError("Field 'email' is required")

    user = User(name=data["name"], email=data["email"])
    db.session.add(user)
    db.session.commit()
    return jsonify(user.to_dict()), 201, {"Location": f"/v1/users/{user.id}"}

Level 3 - Design Challenge

Design and implement a Flask Blueprint for a POST /v1/auth/token endpoint that:

1. Accepts {"email": "...", "password": "..."} in the request body
2. Returns {"access_token": "...", "token_type": "bearer", "expires_in": 3600} on success
3. Returns RFC 7807 errors for: missing fields (422), invalid credentials (401), too many attempts (429)
4. Rate limits to 5 attempts per IP per minute (hint: use a simple in-memory counter)
5. Is fully testable using app.test_client()

Show Reference Solution

# auth/routes.py

import time
import hmac
import hashlib
import jwt as pyjwt
from collections import defaultdict
from flask import Blueprint, request, jsonify, current_app
from ..models import User
from ..exceptions import ValidationError

auth_bp = Blueprint("auth", __name__)

# Simple in-memory rate limiter: {ip: [(timestamp, count)]}
# Production: replace with Redis
_rate_limit_store: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT_WINDOW = 60   # seconds
RATE_LIMIT_MAX = 5       # attempts


def check_rate_limit(ip: str) -> None:
    """Raise 429 if IP has exceeded the rate limit."""
    now = time.monotonic()
    # Keep only timestamps within the window
    _rate_limit_store[ip] = [
        ts for ts in _rate_limit_store[ip]
        if now - ts < RATE_LIMIT_WINDOW
    ]
    if len(_rate_limit_store[ip]) >= RATE_LIMIT_MAX:
        from werkzeug.exceptions import TooManyRequests
        raise TooManyRequests(
            f"Too many login attempts. Try again in {RATE_LIMIT_WINDOW} seconds."
        )
    _rate_limit_store[ip].append(now)


def verify_password(plain: str, hashed: str) -> bool:
    """Constant-time comparison to prevent timing attacks."""
    return hmac.compare_digest(
        hashlib.sha256(plain.encode()).hexdigest(),
        hashed,
    )


def generate_token(user_id: int, secret: str, expires_in: int = 3600) -> str:
    """Generate a JWT access token."""
    import time
    payload = {
        "sub": str(user_id),
        "iat": int(time.time()),
        "exp": int(time.time()) + expires_in,
    }
    return pyjwt.encode(payload, secret, algorithm="HS256")


@auth_bp.route("/token", methods=["POST"])
def create_token():
    """
    Issue a JWT access token for valid credentials.

    Request body: {"email": "...", "password": "..."}
    Returns: {"access_token": "...", "token_type": "bearer", "expires_in": 3600}

    Errors:
        422 - missing required fields
        401 - invalid credentials
        429 - rate limit exceeded (5 attempts per minute per IP)
    """
    ip = request.remote_addr
    check_rate_limit(ip)   # raises 429 if exceeded

    data = request.get_json(silent=True)
    if not data:
        raise ValidationError("Request body must be valid JSON with 'email' and 'password'")

    missing = [f for f in ("email", "password") if not data.get(f)]
    if missing:
        raise ValidationError(
            f"Required fields missing: {', '.join(missing)}",
            missing_fields=missing,
        )

    user = User.query.filter_by(email=data["email"]).first()
    if not user or not verify_password(data["password"], user.password_hash):
        # Generic message - do not reveal whether email exists
        from flask import abort
        abort(401, description="Invalid email or password")

    expires_in = 3600
    token = generate_token(
        user_id=user.id,
        secret=current_app.config["SECRET_KEY"],
        expires_in=expires_in,
    )

    return jsonify({
        "access_token": token,
        "token_type": "bearer",
        "expires_in": expires_in,
    }), 200


# tests/test_auth.py
import pytest
import json

class TestCreateToken:

    def test_valid_credentials_returns_token(self, client, app):
        with app.app_context():
            from app.models import User
            User.create({"email": "[email protected]", "password": "correct-password"})

        response = client.post(
            "/v1/auth/token",
            json={"email": "[email protected]", "password": "correct-password"},
        )

        assert response.status_code == 200
        data = response.get_json()
        assert "access_token" in data
        assert data["token_type"] == "bearer"
        assert data["expires_in"] == 3600

    def test_invalid_password_returns_401(self, client, app):
        with app.app_context():
            from app.models import User
            User.create({"email": "[email protected]", "password": "correct"})

        response = client.post(
            "/v1/auth/token",
            json={"email": "[email protected]", "password": "wrong"},
        )

        assert response.status_code == 401
        data = response.get_json()
        assert data["status"] == 401

    def test_missing_fields_returns_422(self, client):
        response = client.post(
            "/v1/auth/token",
            json={"email": "[email protected]"},
        )

        assert response.status_code == 422
        data = response.get_json()
        assert "missing_fields" in data

    def test_non_json_body_returns_422(self, client):
        response = client.post(
            "/v1/auth/token",
            data="not json",
            content_type="text/plain",
        )
        assert response.status_code == 422

    def test_rate_limit_after_five_attempts(self, client):
        for i in range(5):
            client.post("/v1/auth/token", json={"email": f"a{i}@b.com", "password": "x"})

        response = client.post(
            "/v1/auth/token",
            json={"email": "[email protected]", "password": "x"},
        )
        assert response.status_code == 429

Design decisions:

• hmac.compare_digest prevents timing attacks - an attacker cannot infer whether the email exists by measuring response time
• Rate limiter raises a Werkzeug TooManyRequests (intercepted by HTTPException handler) rather than returning a tuple - consistent with the exception-based error pattern
• Generic 401 message ("Invalid email or password") does not reveal whether the email is registered - user enumeration is a security risk
• expires_in is in the response body (standard OAuth 2.0 convention) so clients know when to refresh the token without decoding the JWT
• In-memory rate limiter is flagged as a prototype - production requires Redis so limits are shared across multiple gunicorn workers

Key Takeaways

• Flask is a WSGI micro-framework: it provides routing and the request/response cycle; everything else is an extension or your responsibility
• Always use the application factory pattern (create_app()): it enables isolated test instances, environment-specific configuration, and avoids circular imports
• request, g, and session are context-local proxies - they look up the current request context on a per-request stack. Accessing them outside a request context raises RuntimeError. Pass data as function arguments to non-view code
• URL converters (<int:id>, <path:p>, <uuid:token>) cast and validate parameters in the route definition - missing or mismatched values return 404 before the view runs
• Use url_for("view_function_name", **kwargs) to generate URLs - it survives URL changes without updating client code
• request.get_json(silent=True) returns None instead of 400 on parse failure - always check for None before accessing fields
• Register a global @app.errorhandler(Exception) handler that logs the traceback and returns a RFC 7807 500 body - never let unhandled exceptions return HTML error pages to API clients
• Blueprints organise routes into modules with a shared URL prefix, registered in create_app() - this is the standard project structure for Flask APIs beyond toy size
• app.test_client() makes the full request/response cycle testable without a running server - use TestConfig with an in-memory SQLite database for fast, isolated tests
• Flask 2.2+ auto-jsonifies dicts; older versions do not - use jsonify() for defensive, version-agnostic code
• Flask is synchronous by default; async def views use a thread executor, not a true event loop - for async-native workloads, use FastAPI with uvicorn

What's Next

Lesson 04 covers FastAPI - the type-annotation-native framework that auto-generates OpenAPI specs, validates request bodies via Pydantic, and runs natively async on uvicorn. You will see how the same REST API you built in Flask looks in FastAPI, why type annotations eliminate entire categories of request.get_json() validation bugs, and when FastAPI's async model genuinely outperforms Flask's threaded model.