← Back to DevBytes
Developer Tutorial July 16, 2026 13 min read

Migrating from Legacy Frameworks to Starlette

What is Starlette and Why Migrate?

Starlette is a lightweight ASGI framework for building high-performance asynchronous web applications and services in Python. It's the foundation upon which FastAPI is built, providing a clean, standards-compliant core that supports HTTP, WebSockets, GraphQL, and background tasks out of the box. Migrating from legacy frameworks like Flask, Django REST Framework, or Pyramid to Starlette unlocks significant performance gains, modern async capabilities, and a simpler deployment model.

The Starlette Advantage

Legacy WSGI frameworks operate synchronously, handling one request per worker thread or process. This model struggles under high concurrency, especially with I/O-bound workloads like database queries or external API calls. Starlette, built on ASGI, uses an event loop to handle thousands of concurrent connections with minimal resource overhead. Key benefits include:

Common Legacy Frameworks

This tutorial covers migration patterns for three widely-used legacy frameworks:

Getting Started with Migration

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

Project Setup

Install Starlette with an ASGI server. Uvicorn is the recommended choice for production and development:

pip install starlette uvicorn[standard]
# Optional but recommended for type-safe requests/responses
pip install python-multipart  # for form parsing
pip install httpx             # for async testing

Create a minimal application file to verify your setup:

# app.py
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    return JSONResponse({"hello": "world"})

routes = [
    Route("/", endpoint=homepage),
]

app = Starlette(debug=True, routes=routes)

Run with Uvicorn:

uvicorn app:app --reload --host 0.0.0.0 --port 8000

Basic Route Migration: Flask to Starlette

Consider a typical Flask route with query parameters and a JSON response:

# Legacy Flask route
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/api/users", methods=["GET"])
def list_users():
    page = request.args.get("page", 1, type=int)
    limit = request.args.get("limit", 20, type=int)
    users = fetch_users_sync(page, limit)  # synchronous DB call
    return jsonify({"users": users, "page": page})

The Starlette equivalent uses async handlers and explicit parameter extraction:

# Starlette migration
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

async def list_users(request: Request):
    page = int(request.query_params.get("page", 1))
    limit = int(request.query_params.get("limit", 20))
    users = await fetch_users_async(page, limit)  # async DB call
    return JSONResponse({"users": users, "page": page})

routes = [Route("/api/users", endpoint=list_users)]
app = Starlette(routes=routes)

Path Parameters and Type Conversion

Flask uses angle-bracket converters. Starlette uses curly-brace parameters with optional type casting:

# Flask: path with type converter
@app.route("/users/<int:user_id>")
def get_user(user_id):
    ...

# Starlette: path parameter with type annotation
from starlette.routing import Route

async def get_user(request: Request):
    user_id = int(request.path_params["user_id"])
    ...

# Route definition uses {param:type} syntax
routes = [Route("/users/{user_id:int}", endpoint=get_user)]

Async Database Integration

One of the biggest wins when migrating is switching to async database drivers. Here's how to set up connection pooling with asyncpg for PostgreSQL:

# database.py
import asyncpg
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

async def create_pool():
    return await asyncpg.create_pool(
        host="localhost",
        database="mydb",
        user="myuser",
        password="mypassword",
        min_size=5,
        max_size=20
    )

async def get_user_by_id(request: Request):
    pool = request.app.state.pool
    user_id = int(request.path_params["user_id"])
    async with pool.acquire() as connection:
        row = await connection.fetchrow(
            "SELECT id, name, email FROM users WHERE id = $1", user_id
        )
    if row is None:
        return JSONResponse({"error": "Not found"}, status_code=404)
    return JSONResponse(dict(row))

routes = [Route("/users/{user_id:int}", endpoint=get_user_by_id)]

app = Starlette(routes=routes)

@app.on_event("startup")
async def startup():
    app.state.pool = await create_pool()

@app.on_event("shutdown")
async def shutdown():
    await app.state.pool.close()

Middleware Migration

Flask middleware often uses the before_request / after_request hooks or WSGI middleware wrappers. Starlette uses a clean, ASGI-native middleware stack:

# Flask middleware pattern
@app.before_request
def authenticate():
    token = request.headers.get("Authorization")
    if token:
        g.current_user = verify_token(token)

# Starlette middleware as a class
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class AuthenticationMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        token = request.headers.get("Authorization")
        if token:
            request.state.user = await verify_token_async(token)
        else:
            request.state.user = None
        response = await call_next(request)
        return response

app = Starlette(
    routes=routes,
    middleware=[
        Middleware(AuthenticationMiddleware),
        Middleware(CORSMiddleware, allow_origins=["*"]),
    ]
)

For simple header manipulation or logging, you can write a pure ASGI middleware function without the class overhead:

# Functional ASGI middleware
async def add_timestamp_middleware(scope, receive, send):
    if scope["type"] != "http":
        # Pass through non-HTTP scopes (like websocket lifespan)
        await app(scope, receive, send)
        return
    async def send_wrapper(message):
        if message["type"] == "http.response.start":
            headers = dict(message.get("headers", []))
            headers[b"x-response-time"] = str(time.time()).encode()
            message["headers"] = list(headers.items())
        await send(message)
    await app(scope, receive, send_wrapper)

WebSocket Support

This is where Starlette truly outshines legacy frameworks. Adding WebSocket support in Flask requires extensions like Flask-SocketIO with additional infrastructure. Starlette handles WebSockets natively:

# Starlette WebSocket endpoint
from starlette.applications import Starlette
from starlette.routing import WebSocketRoute
from starlette.websockets import WebSocket

async def chat_endpoint(websocket: WebSocket):
    await websocket.accept()
    # Send a welcome message
    await websocket.send_json({"type": "welcome", "message": "Connected to chat"})
    
    while True:
        try:
            data = await websocket.receive_json()
            if data["type"] == "message":
                # Echo back or broadcast
                await websocket.send_json({
                    "type": "response",
                    "echo": data["content"]
                })
            elif data["type"] == "ping":
                await websocket.send_json({"type": "pong"})
        except WebSocketDisconnect:
            print("Client disconnected")
            break

routes = [
    WebSocketRoute("/ws/chat", endpoint=chat_endpoint),
]

app = Starlette(routes=routes)

Step-by-Step Migration Strategy

Phase 1: Assessment and Inventory

Before writing any code, catalog your existing endpoints, middleware, and dependencies:

Phase 2: Parallel Development with API Gateway

Run Starlette alongside your legacy application behind a reverse proxy or API gateway. This allows gradual cutover without downtime:

# nginx.conf example for parallel deployment
upstream legacy_api {
    server 127.0.0.1:5000;  # Flask/Werkzeug
}
upstream starlette_api {
    server 127.0.0.1:8000;  # Uvicorn
}

server {
    listen 80;
    location /api/v1/ {
        proxy_pass http://legacy_api;  # Old endpoints
    }
    location /api/v2/ {
        proxy_pass http://starlette_api;  # New Starlette endpoints
    }
    location /ws/ {
        proxy_pass http://starlette_api;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

Phase 3: Incremental Cutover

Migrate endpoints in order of complexity, starting with the simplest read-only GET routes:

Practical Migration Examples

Flask Route with Validation to Starlette

A common Flask pattern uses decorator-based validation. In Starlette, you can use dependency injection or middleware:

# Flask with marshmallow validation
from flask import request, jsonify
from marshmallow import Schema, fields, ValidationError

class UserSchema(Schema):
    name = fields.Str(required=True)
    email = fields.Email(required=True)

@app.route("/users", methods=["POST"])
def create_user():
    try:
        data = UserSchema().load(request.json)
    except ValidationError as err:
        return jsonify({"errors": err.messages}), 400
    user = create_user_sync(data)
    return jsonify(user), 201

# Starlette equivalent with manual validation
from starlette.requests import Request
from starlette.responses import JSONResponse

class UserSchema(Schema):
    name = fields.Str(required=True)
    email = fields.Email(required=True)

async def create_user(request: Request):
    body = await request.json()
    try:
        validated_data = UserSchema().load(body)
    except ValidationError as err:
        return JSONResponse({"errors": err.messages}, status_code=400)
    
    user = await create_user_async(validated_data)
    return JSONResponse(user, status_code=201)

Django REST Framework ViewSet to Starlette

DRF ViewSets pack list, create, retrieve, update, and destroy into one class. In Starlette, you define separate async functions for each operation:

# DRF ViewSet (legacy)
from rest_framework import viewsets
from rest_framework.response import Response

class ArticleViewSet(viewsets.ModelViewSet):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer

    def list(self, request):
        articles = self.get_queryset()
        serializer = self.get_serializer(articles, many=True)
        return Response(serializer.data)

# Starlette equivalent with separate routes
from starlette.routing import Route, Router

async def list_articles(request: Request):
    pool = request.app.state.pool
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT * FROM articles ORDER BY created_at DESC")
    articles = [dict(row) for row in rows]
    return JSONResponse(articles)

async def get_article(request: Request):
    article_id = int(request.path_params["article_id"])
    pool = request.app.state.pool
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT * FROM articles WHERE id = $1", article_id
        )
    if not row:
        return JSONResponse({"detail": "Not found"}, status_code=404)
    return JSONResponse(dict(row))

async def create_article(request: Request):
    body = await request.json()
    pool = request.app.state.pool
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            """INSERT INTO articles (title, content, author_id) 
               VALUES ($1, $2, $3) RETURNING *""",
            body["title"], body["content"], request.state.user["id"]
        )
    return JSONResponse(dict(row), status_code=201)

router = Router([
    Route("/articles", endpoint=list_articles, methods=["GET"]),
    Route("/articles", endpoint=create_article, methods=["POST"]),
    Route("/articles/{article_id:int}", endpoint=get_article, methods=["GET"]),
])

Error Handling Migration

Flask uses errorhandler decorators. Starlette provides a centralized exception handling system:

# Flask error handlers
@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "Resource not found"}), 404

@app.errorhandler(ValueError)
def handle_value_error(error):
    return jsonify({"error": str(error)}), 400

# Starlette exception handlers
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse

async def not_found_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        {"error": "Resource not found", "path": str(request.url)},
        status_code=404
    )

async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        {"error": str(exc), "type": "validation_error"},
        status_code=400
    )

exception_handlers = {
    404: not_found_handler,
    ValueError: value_error_handler,
}

app = Starlette(
    routes=routes,
    exception_handlers=exception_handlers,
)

Dependency Injection Pattern

Starlette doesn't include a built-in dependency injection system like FastAPI, but you can implement a clean pattern using closures or class-based views:

# Dependency injection via factory functions
from typing import Callable
from starlette.requests import Request
from starlette.responses import JSONResponse

def require_auth(endpoint: Callable):
    """Decorator-like dependency injection for auth"""
    async def wrapper(request: Request):
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return JSONResponse({"error": "Unauthorized"}, status_code=401)
        token = auth_header.split(" ")[1]
        user = await verify_token(token)
        if not user:
            return JSONResponse({"error": "Invalid token"}, status_code=401)
        request.state.user = user
        return await endpoint(request)
    return wrapper

async def protected_dashboard(request: Request):
    user = request.state.user
    return JSONResponse({"message": f"Welcome {user['name']}"})

routes = [
    Route("/dashboard", endpoint=require_auth(protected_dashboard)),
]

For more complex dependencies, use a class-based approach:

# Class-based dependency container
class Dependencies:
    def __init__(self, pool, redis_client, config):
        self.pool = pool
        self.redis = redis_client
        self.config = config
    
    async def get_user_repository(self):
        return UserRepository(self.pool)
    
    async def get_cache(self):
        return CacheService(self.redis)

async def setup_dependencies(app: Starlette):
    pool = await create_pool()
    redis = await create_redis_client()
    deps = Dependencies(pool, redis, config=app.state.config)
    app.state.dependencies = deps

@app.on_event("startup")
async def startup():
    await setup_dependencies(app)

Streaming Responses and Server-Sent Events

Starlette supports streaming responses natively, which is useful for large file downloads or real-time event streams:

# Flask streaming (sync generator)
from flask import Response, stream_with_context
import csv, io

@app.route("/export/users.csv")
def export_users():
    def generate():
        writer = csv.writer(io.StringIO())
        writer.writerow(["id", "name", "email"])
        for user in fetch_users_chunked():
            writer.writerow([user.id, user.name, user.email])
            yield io.getvalue()
    return Response(stream_with_context(generate()), mimetype="text/csv")

# Starlette streaming (async generator)
from starlette.responses import StreamingResponse
import csv, io

async def export_users(request: Request):
    async def generate():
        buffer = io.StringIO()
        writer = csv.writer(buffer)
        writer.writerow(["id", "name", "email"])
        yield buffer.getvalue()
        buffer.seek(0)
        buffer.truncate(0)
        
        async for user in fetch_users_async_chunked():
            writer.writerow([user["id"], user["name"], user["email"]])
            yield buffer.getvalue()
            buffer.seek(0)
            buffer.truncate(0)
    
    return StreamingResponse(
        generate(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=users.csv"}
    )

File Upload Migration

Flask uses request.files. Starlette handles multipart form data through request.form():

# Flask file upload
@app.route("/upload", methods=["POST"])
def upload_file():
    file = request.files["file"]
    filename = secure_filename(file.filename)
    file.save(os.path.join(UPLOAD_DIR, filename))
    return jsonify({"filename": filename})

# Starlette file upload
from starlette.form_data import UploadFile
import aiofiles

async def upload_file(request: Request):
    form = await request.form()
    uploaded_file: UploadFile = form["file"]
    filename = uploaded_file.filename
    
    # Async file writing
    async with aiofiles.open(f"uploads/{filename}", "wb") as f:
        while True:
            chunk = await uploaded_file.read(8192)  # 8KB chunks
            if not chunk:
                break
            await f.write(chunk)
    
    return JSONResponse({"filename": filename, "size": uploaded_file.size})

Best Practices for Migration

1. Keep the Database Layer Async-Native

The biggest performance regression comes from mixing sync and async code. Use async database drivers (asyncpg, aiomysql, aiosqlite) rather than wrapping synchronous ORMs in threads:

# DON'T: Run sync ORM in thread pool
import asyncio
from sqlalchemy.orm import Session

async def get_users(request):
    loop = asyncio.get_event_loop()
    users = await loop.run_in_executor(None, sync_orm_query)  # Blocks a thread
    return JSONResponse(users)

# DO: Use native async drivers
async def get_users(request):
    async with request.app.state.pool.acquire() as conn:
        rows = await conn.fetch("SELECT * FROM users")  # True async
    return JSONResponse([dict(r) for r in rows])

2. Use Lifespan Events for Resource Management

Starlette's lifespan system is superior to scattered startup/shutdown decorators. Use it for connection pools, cache clients, and background task schedulers:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: Starlette):
    # Startup
    pool = await asyncpg.create_pool(...)
    redis = await aioredis.create_redis_pool(...)
    app.state.pool = pool
    app.state.redis = redis
    print("Resources initialized")
    
    yield  # Application runs here
    
    # Shutdown
    await pool.close()
    redis.close()
    await redis.wait_closed()
    print("Resources cleaned up")

app = Starlette(routes=routes, lifespan=lifespan)

3. Structure Routes with Router Composition

For larger applications, use Router objects to organize routes by domain:

from starlette.routing import Router, Mount

# users/endpoints.py
users_router = Router([
    Route("/", endpoint=list_users, methods=["GET"]),
    Route("/{user_id:int}", endpoint=get_user, methods=["GET"]),
    Route("/", endpoint=create_user, methods=["POST"]),
])

# articles/endpoints.py
articles_router = Router([
    Route("/", endpoint=list_articles, methods=["GET"]),
    Route("/{article_id:int}", endpoint=get_article, methods=["GET"]),
])

# Main application
app = Starlette(routes=[
    Mount("/api/users", app=users_router),
    Mount("/api/articles", app=articles_router),
])

4. Test with Async Test Clients

Use httpx or Starlette's built-in TestClient (which wraps httpx) for async testing:

# tests/test_users.py
from starlette.testclient import TestClient
from app import app
import pytest

def test_list_users():
    with TestClient(app) as client:
        response = client.get("/api/users?page=1&limit=10")
        assert response.status_code == 200
        assert "users" in response.json()

# Async test with httpx directly
import httpx
import pytest

@pytest.mark.asyncio
async def test_create_user_async():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/api/users",
            json={"name": "Alice", "email": "alice@example.com"}
        )
    assert response.status_code == 201

5. Handle Backwards Compatibility Gracefully

During migration, maintain API compatibility. Use middleware to transform responses if needed:

# Compatibility middleware for legacy clients
class LegacyResponseAdapter(BaseHTTPMiddleware):
    """Wraps new response format in legacy wrapper for old clients"""
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        if request.headers.get("X-API-Version") == "v1":
            body = await response.body()
            data = json.loads(body)
            wrapped = {"status": "ok", "result": data}  # Legacy wrapper
            return JSONResponse(wrapped, status_code=response.status_code)
        return response

6. Monitor and Log with Structuring

Add structured logging middleware to track migration progress and catch regressions:

import logging
import time
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger("migration")

class MigrationMetricsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        elapsed = time.perf_counter() - start
        
        logger.info({
            "path": str(request.url.path),
            "method": request.method,
            "status": response.status_code,
            "duration_ms": round(elapsed * 1000, 2),
            "framework": "starlette",
        })
        return response

7. Graceful Degradation for Async Dependencies

Not all libraries are async yet. For unavoidable sync operations, use run_in_executor judiciously with a dedicated thread pool:

from concurrent.futures import ThreadPoolExecutor
import asyncio

# Create a bounded executor to avoid thread exhaustion
cpu_bound_executor = ThreadPoolExecutor(max_workers=4)

async def generate_pdf_report(request: Request):
    report_id = request.path_params["report_id"]
    loop = asyncio.get_running_loop()
    
    # Sync PDF generation library
    pdf_bytes = await loop.run_in_executor(
        cpu_bound_executor,
        sync_pdf_library.generate,  # blocking call
        report_id
    )
    
    return Response(pdf_bytes, media_type="application/pdf")

Common Pitfalls and Solutions

Pitfall: Blocking the Event Loop

The most frequent migration mistake is calling synchronous code inside async handlers without delegation to a thread pool. This blocks the entire event loop:

# WRONG: Blocks all other requests
async def bad_handler(request):
    result = requests.get("https://api.external.com/data")  # Synchronous HTTP call!
    return JSONResponse(result.json())

# CORRECT: Use httpx for async HTTP
async def good_handler(request):
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.external.com/data")
    return JSONResponse(response.json())

Pitfall: Losing Request Context

Flask's g object is thread-local. Starlette uses request.state which is request-scoped and async-safe:

# Flask pattern (won't work in async)
from flask import g

@app.before_request
def load_user():
    g.user = get_user_from_session()

# Starlette pattern (correct)
class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request.state.user = await get_user_from_session(request)
        return await call_next(request)

Pitfall: Mixing Async and Sync Middleware

WSGI middleware cannot be directly used with ASGI. Always rewrite middleware as native ASGI:

# DON'T attempt to wrap WSGI middleware
# from flask_cors import cross_origin  # Won't work

# DO: Use ASGI-native CORS middleware
from starlette.middleware.cors import CORSMiddleware

app = Starlette(
    middleware=[
        Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"]),
    ]
)

Conclusion

Migrating from legacy frameworks to Starlette is a strategic investment that pays dividends in performance, scalability, and developer experience. The async-first architecture eliminates thread pool bottlenecks, while the clean ASGI foundation opens doors to WebSockets, server-sent events, and modern deployment patterns. By following the phased migration approach — assessing dependencies, running parallel deployments, and incrementally cutting over endpoints — teams can transition with minimal risk and zero downtime. The key to success lies in embracing async-native libraries throughout the stack, restructuring middleware as ASGI components, and using Starlette's lifespan events for clean resource management. With the patterns and examples provided in this tutorial, you have a complete roadmap for transforming synchronous legacy APIs into high-performance, modern async services built on Starlette.

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles