Falcon REST Framework — Deep Dive

Production Falcon architecture: custom middleware chains, ASGI WebSockets, media handlers, hooks, and scaling strategies for high-throughput APIs.

Application structure

A well-organized Falcon application separates resources, middleware, and configuration:

# app.py
import falcon
import falcon.asgi

from resources.users import UserResource, UserCollectionResource
from resources.health import HealthResource
from middleware.auth import AuthMiddleware
from middleware.logging import RequestLoggingMiddleware
from middleware.cors import CORSMiddleware

def create_app():
    app = falcon.asgi.App(
        middleware=[
            CORSMiddleware(),
            RequestLoggingMiddleware(),
            AuthMiddleware(),
        ]
    )
    
    app.add_route("/health", HealthResource())
    app.add_route("/api/v1/users", UserCollectionResource())
    app.add_route("/api/v1/users/{user_id:int}", UserResource())
    
    return app

app = create_app()

Resource design patterns

Complex resources benefit from separation between collection and item endpoints:

class UserCollectionResource:
    async def on_get(self, req, resp):
        """List users with pagination."""
        page = int(req.get_param("page") or 1)
        per_page = min(int(req.get_param("per_page") or 20), 100)
        offset = (page - 1) * per_page
        
        users = await db.fetch_users(limit=per_page, offset=offset)
        total = await db.count_users()
        
        resp.media = {
            "data": [u.to_dict() for u in users],
            "pagination": {
                "page": page,
                "per_page": per_page,
                "total": total,
                "pages": (total + per_page - 1) // per_page,
            }
        }
    
    async def on_post(self, req, resp):
        """Create a new user."""
        data = await req.get_media()
        
        # Manual validation
        errors = validate_user_input(data)
        if errors:
            resp.status = falcon.HTTP_422
            resp.media = {"errors": errors}
            return
        
        user = await db.create_user(data)
        resp.status = falcon.HTTP_201
        resp.location = f"/api/v1/users/{user.id}"
        resp.media = user.to_dict()


class UserResource:
    async def on_get(self, req, resp, user_id):
        user = await db.get_user(user_id)
        if not user:
            raise falcon.HTTPNotFound(
                title="User not found",
                description=f"No user with id {user_id}"
            )
        resp.media = user.to_dict()
    
    async def on_patch(self, req, resp, user_id):
        """Partial update."""
        data = await req.get_media()
        user = await db.update_user(user_id, data)
        if not user:
            raise falcon.HTTPNotFound()
        resp.media = user.to_dict()
    
    async def on_delete(self, req, resp, user_id):
        deleted = await db.delete_user(user_id)
        if not deleted:
            raise falcon.HTTPNotFound()
        resp.status = falcon.HTTP_204

Hooks for per-route logic

Falcon hooks provide before/after logic at the resource or responder level:

import falcon

def require_admin(req, resp, resource, params):
    """Before hook: check admin permissions."""
    if not req.context.get("user", {}).get("is_admin"):
        raise falcon.HTTPForbidden(
            description="Admin access required"
        )

def log_modification(req, resp, resource):
    """After hook: audit log for write operations."""
    audit_logger.info(
        f"Modified by {req.context.get('user_id')}: "
        f"{req.method} {req.path} -> {resp.status}"
    )

class AdminResource:
    @falcon.before(require_admin)
    @falcon.after(log_modification)
    async def on_delete(self, req, resp, resource_id):
        await db.delete_resource(resource_id)
        resp.status = falcon.HTTP_204
    
    @falcon.before(require_admin)
    async def on_put(self, req, resp, resource_id):
        data = await req.get_media()
        await db.update_resource(resource_id, data)
        resp.media = {"updated": True}

Hooks are more granular than middleware — they apply to specific responder methods, not all requests.

Custom media handlers

Falcon uses media handlers for serialization and deserialization. The default handles JSON, but you can add custom formats:

import msgpack
import falcon

class MessagePackHandler(falcon.media.BaseHandler):
    def serialize(self, media, content_type):
        return msgpack.packb(media, use_bin_type=True)
    
    def deserialize(self, stream, content_type, content_length):
        data = stream.read()
        return msgpack.unpackb(data, raw=False)

app = falcon.asgi.App()

extra_handlers = {
    "application/msgpack": MessagePackHandler(),
    "application/x-msgpack": MessagePackHandler(),
}
app.req_options.media_handlers.update(extra_handlers)
app.resp_options.media_handlers.update(extra_handlers)

Clients can then request MessagePack format via the Accept header, and Falcon will automatically use the right serializer. This is valuable for services where JSON parsing is a performance bottleneck.

Middleware deep dive

The three-phase middleware model provides precise control:

class RequestContextMiddleware:
    async def process_request(self, req, resp):
        """Phase 1: Before routing. Set up request context."""
        req.context.request_id = str(uuid.uuid4())
        req.context.start_time = time.perf_counter()
    
    async def process_resource(self, req, resp, resource, params):
        """Phase 2: After routing, before responder.
        'resource' is the matched resource instance.
        'params' are the extracted URL parameters.
        """
        req.context.resource_name = resource.__class__.__name__
        
        # Convert string params to proper types
        if "user_id" in params:
            params["user_id"] = int(params["user_id"])
    
    async def process_response(self, req, resp, resource, req_succeeded):
        """Phase 3: After responder (or error).
        'req_succeeded' is False if an exception was raised.
        """
        elapsed = time.perf_counter() - req.context.start_time
        resp.set_header("X-Request-ID", req.context.request_id)
        resp.set_header("X-Response-Time", f"{elapsed:.4f}")
        
        metrics.histogram(
            "http_request_duration",
            elapsed,
            tags={
                "method": req.method,
                "path": req.path,
                "status": resp.status[:3],
                "success": str(req_succeeded),
            }
        )

Error handling

Falcon provides typed HTTP exceptions and custom error serialization:

class CustomErrorHandler:
    @staticmethod
    async def handle(req, resp, ex, params):
        """Global error handler."""
        if isinstance(ex, falcon.HTTPError):
            resp.status = ex.status
            resp.media = {
                "error": {
                    "status": ex.status,
                    "title": ex.title or "Error",
                    "description": ex.description or "",
                }
            }
        else:
            logger.exception("Unhandled error", exc_info=ex)
            resp.status = falcon.HTTP_500
            resp.media = {
                "error": {
                    "status": "500",
                    "title": "Internal Server Error",
                    "description": "An unexpected error occurred",
                }
            }

app = falcon.asgi.App()
app.add_error_handler(Exception, CustomErrorHandler.handle)

ASGI WebSocket support

Falcon’s ASGI mode includes WebSocket handling:

class NotificationWebSocket:
    async def on_websocket(self, req, ws):
        try:
            await ws.accept()
            
            user_id = req.get_param("user_id")
            if not user_id:
                await ws.close(1008)
                return
            
            # Subscribe to notifications
            async with notification_bus.subscribe(user_id) as subscription:
                async for notification in subscription:
                    await ws.send_media(notification.to_dict())
        
        except falcon.WebSocketDisconnected:
            pass
        finally:
            await notification_bus.unsubscribe(user_id)

app.add_route("/ws/notifications", NotificationWebSocket())

Streaming responses

For large payloads, Falcon supports streaming in ASGI mode:

class ExportResource:
    async def on_get(self, req, resp):
        resp.content_type = "application/jsonl"
        resp.stream = self._generate_export()
    
    async def _generate_export(self):
        async for batch in db.fetch_all_batches(size=1000):
            for record in batch:
                yield json.dumps(record).encode() + b"\n"

Testing

Falcon provides a testing client that doesn’t require a running server:

from falcon import testing
import pytest

@pytest.fixture
def client():
    return testing.TestClient(create_app())

def test_list_users(client):
    result = client.simulate_get("/api/v1/users")
    assert result.status == falcon.HTTP_200
    data = result.json
    assert "data" in data
    assert "pagination" in data

def test_create_user(client):
    body = {"name": "Alice", "email": "alice@example.com"}
    result = client.simulate_post("/api/v1/users", json=body)
    assert result.status == falcon.HTTP_201
    assert "id" in result.json

def test_not_found(client):
    result = client.simulate_get("/api/v1/users/99999")
    assert result.status == falcon.HTTP_404

The simulate_* methods bypass the network stack entirely, making tests fast and deterministic.

Deployment and scaling

# ASGI deployment with uvicorn
uvicorn app:app --workers 4 --host 0.0.0.0 --port 8000

# WSGI deployment with gunicorn
gunicorn app:app -w 4 -b 0.0.0.0:8000 --preload

The --preload flag with gunicorn loads the app before forking workers, reducing memory usage through copy-on-write.

For high-throughput scenarios:

  • Use orjson as the JSON handler (3-10x faster than stdlib json)
  • Enable response caching at the middleware level for read-heavy endpoints
  • Use connection pooling for all external dependencies
  • Profile with py-spy to find actual bottlenecks rather than guessing
# Switch to orjson for faster JSON
import orjson

class ORJSONHandler(falcon.media.BaseHandler):
    def serialize(self, media, content_type):
        return orjson.dumps(media)
    
    def deserialize(self, stream, content_type, content_length):
        return orjson.loads(stream.read())

app.req_options.media_handlers["application/json"] = ORJSONHandler()
app.resp_options.media_handlers["application/json"] = ORJSONHandler()

The one thing to remember: Falcon’s three-phase middleware, resource-oriented routing, pluggable media handlers, and minimal per-request overhead create an architecture where every microsecond is intentional — making it the Python framework to reach for when API throughput and latency are non-negotiable requirements.

pythonweb-frameworksrest-apifalcon

See Also

  • Python Aiohttp Client Understand Aiohttp Client through a practical analogy so your Python decisions become faster and clearer.
  • Python Api Client Design Why building your own API client in Python is like creating a TV remote that only has the buttons you actually need.
  • Python Api Documentation Swagger Swagger turns your Python API into an interactive playground where anyone can click buttons to try it out — no coding required.
  • Python Api Mocking Responses Why testing with fake API responses is like rehearsing a play with stand-ins before the real actors show up.
  • Python Api Pagination Clients Why APIs send data in pages, and how Python handles it — like reading a book one chapter at a time instead of swallowing the whole thing.

Related Topics

ab-testing How tech companies run thousands of experiments at once to improve their products — the scientific method applied to everything from button colors to recommendation algorithms. 3 levels → activation-functions Why neural networks need these tiny mathematical functions — and how ReLU's simplicity accidentally made deep learning possible. 3 levels → ai-agents-architecture How AI systems go from answering questions to actually doing things — the design patterns that turn language models into autonomous agents that browse, code, and plan. 3 levels →