← Back to DevBytes

Migrating from Flask to FastAPI: Step-by-Step Guide

Understanding the Migration from Flask to FastAPI

Flask has been the beloved micro-framework for Python web applications for over a decade. Its minimalist philosophy, straightforward routing, and extensive plugin ecosystem made it the default choice for countless projects. However, as modern API development demands have evolved—faster performance, built-in data validation, auto-generated documentation, and native async support—FastAPI has emerged as a compelling alternative. Migrating from Flask to FastAPI is not about abandoning a framework; it's about strategically upgrading your application to leverage these modern capabilities while preserving your existing business logic.

This tutorial provides a complete, practical, step-by-step guide for developers undertaking this migration. You'll learn how to map Flask concepts to their FastAPI equivalents, refactor route handlers, introduce Pydantic models for robust validation, handle async operations, and update your testing suite—all with fully executable code examples.

Why Migrate from Flask to FastAPI?

🚀 Deploy your AI agent in 10 minutes

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

Try it free →

Understanding the concrete benefits helps justify the migration effort:

Key Conceptual Differences Between Flask and FastAPI

Before diving into code, let's map the core concepts side by side. This mental model will make the migration process intuitive:

Application and Routing

Request Data Access

Response Handling

Middleware

Error Handling

Step-by-Step Migration Guide

Let's walk through a real migration. We'll start with a typical Flask application for a simple task management API and transform it into an equivalent FastAPI application. All code examples are complete and ready to run.

Step 1: Setting Up the FastAPI Project Environment

First, install the required dependencies. FastAPI requires an ASGI server (Uvicorn) and optionally an async database driver:

pip install fastapi uvicorn
pip install pydantic
# For async database support (example uses databases library)
pip install databases[sqlite]

Create a new project structure. Unlike Flask, which often uses a single app.py file, FastAPI benefits from a modular layout from the start:

project/
├── main.py          # Application entry point and route definitions
├── models.py        # Pydantic models for request/response schemas
├── database.py      # Database connection and queries
├── dependencies.py  # Shared dependency injection functions
└── tests/
    └── test_api.py  # Updated test suite

Step 2: Converting Flask Routes to FastAPI Endpoints

Consider this original Flask application with CRUD endpoints for tasks:

# Original Flask app (app.py)
from flask import Flask, request, jsonify, abort
from flask_sqlalchemy import SQLAlchemy
import sqlite3

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///tasks.db'
db = SQLAlchemy(app)

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    completed = db.Column(db.Boolean, default=False)

@app.route('/tasks', methods=['GET'])
def get_tasks():
    tasks = Task.query.all()
    return jsonify([{'id': t.id, 'title': t.title, 'completed': t.completed} for t in tasks])

@app.route('/tasks', methods=['POST'])
def create_task():
    data = request.get_json()
    if not data or 'title' not in data:
        abort(400, description="Title is required")
    task = Task(title=data['title'], completed=data.get('completed', False))
    db.session.add(task)
    db.session.commit()
    return jsonify({'id': task.id, 'title': task.title, 'completed': task.completed}), 201

@app.route('/tasks/<int:task_id>', methods=['GET'])
def get_task(task_id):
    task = Task.query.get(task_id)
    if not task:
        abort(404, description="Task not found")
    return jsonify({'id': task.id, 'title': task.title, 'completed': task.completed})

@app.route('/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
    task = Task.query.get(task_id)
    if not task:
        abort(404, description="Task not found")
    data = request.get_json()
    task.title = data.get('title', task.title)
    task.completed = data.get('completed', task.completed)
    db.session.commit()
    return jsonify({'id': task.id, 'title': task.title, 'completed': task.completed})

@app.route('/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    task = Task.query.get(task_id)
    if not task:
        abort(404, description="Task not found")
    db.session.delete(task)
    db.session.commit()
    return '', 204

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

Now, let's convert this to FastAPI. Notice how each Flask @app.route with a methods parameter becomes a distinct FastAPI decorator:

# FastAPI main.py
from fastapi import FastAPI, HTTPException, status, Depends
from typing import List
import models
import database
from dependencies import get_db

app = FastAPI(
    title="Task Management API",
    description="Migrated from Flask to FastAPI",
    version="1.0.0"
)

@app.get("/tasks", response_model=List[models.TaskResponse])
async def get_tasks(db=Depends(get_db)):
    """Retrieve all tasks from the database."""
    tasks = await db.fetch_all("SELECT id, title, completed FROM tasks")
    return [dict(task) for task in tasks]

@app.post("/tasks", response_model=models.TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(task_create: models.TaskCreate, db=Depends(get_db)):
    """Create a new task. Title is required."""
    query = "INSERT INTO tasks (title, completed) VALUES (:title, :completed)"
    task_id = await db.execute(query, values={"title": task_create.title, "completed": task_create.completed})
    created_task = await db.fetch_one("SELECT id, title, completed FROM tasks WHERE id = :id", {"id": task_id})
    return dict(created_task)

@app.get("/tasks/{task_id}", response_model=models.TaskResponse)
async def get_task(task_id: int, db=Depends(get_db)):
    """Retrieve a single task by its ID."""
    task = await db.fetch_one("SELECT id, title, completed FROM tasks WHERE id = :id", {"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return dict(task)

@app.put("/tasks/{task_id}", response_model=models.TaskResponse)
async def update_task(task_id: int, task_update: models.TaskUpdate, db=Depends(get_db)):
    """Update an existing task's title and/or completion status."""
    task = await db.fetch_one("SELECT id, title, completed FROM tasks WHERE id = :id", {"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    update_values = task_update.model_dump(exclude_unset=True)
    if update_values:
        set_clause = ", ".join(f"{key} = :{key}" for key in update_values)
        await db.execute(f"UPDATE tasks SET {set_clause} WHERE id = :id", {**update_values, "id": task_id})
    updated_task = await db.fetch_one("SELECT id, title, completed FROM tasks WHERE id = :id", {"id": task_id})
    return dict(updated_task)

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int, db=Depends(get_db)):
    """Delete a task by its ID."""
    task = await db.fetch_one("SELECT id FROM tasks WHERE id = :id", {"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    await db.execute("DELETE FROM tasks WHERE id = :id", {"id": task_id})
    # Returning None with 204 status code

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Key changes to observe:

Step 3: Handling Request Data with Pydantic Models

In Flask, request validation is manual and error-prone. FastAPI uses Pydantic models to define exactly what data your endpoints expect and return. Create a models.py file:

# models.py
from pydantic import BaseModel, Field
from typing import Optional

class TaskCreate(BaseModel):
    """Schema for creating a new task."""
    title: str = Field(..., min_length=1, max_length=100, description="Task title")
    completed: bool = Field(False, description="Whether the task is completed")

class TaskUpdate(BaseModel):
    """Schema for updating an existing task. All fields are optional."""
    title: Optional[str] = Field(None, min_length=1, max_length=100)
    completed: Optional[bool] = None

class TaskResponse(BaseModel):
    """Schema for task data returned to the client."""
    id: int
    title: str
    completed: bool

    class Config:
        # Enables population from ORM objects or dicts
        from_attributes = True

When a request arrives, FastAPI automatically parses the JSON body into the Pydantic model, validates all fields, and provides clear error messages for invalid data. Compare this to Flask, where you'd write manual if not data or 'title' not in data checks scattered across endpoints. With FastAPI, validation is centralized in the model definitions.

Step 4: Migrating Database Operations to Async

Flask typically uses synchronous ORMs like SQLAlchemy. FastAPI's async capabilities shine when paired with async database libraries. Here's a complete database.py using the databases library with SQLite:

# database.py
import databases
import sqlalchemy

DATABASE_URL = "sqlite:///tasks.db"

# SQLAlchemy metadata for table definitions
metadata = sqlalchemy.MetaData()

tasks_table = sqlalchemy.Table(
    "tasks",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True),
    sqlalchemy.Column("title", sqlalchemy.String(100), nullable=False),
    sqlalchemy.Column("completed", sqlalchemy.Boolean, default=False),
)

# Async database instance
database = databases.Database(DATABASE_URL)

async def create_tables():
    """Create tables on application startup."""
    engine = sqlalchemy.create_engine(
        DATABASE_URL, connect_args={"check_same_thread": False}
    )
    metadata.create_all(engine)
    engine.dispose()

async def connect_database():
    """Connect to database on application startup."""
    await database.connect()

async def disconnect_database():
    """Disconnect from database on application shutdown."""
    await database.disconnect()

Now create dependencies.py for FastAPI's dependency injection system. This replaces Flask's global db object pattern:

# dependencies.py
from database import database

async def get_db():
    """Dependency that provides a database connection for each request.
    
    FastAPI will call this function for every route that declares
    `db = Depends(get_db)`. The connection is automatically managed.
    """
    return database

Wire everything together in main.py with startup and shutdown event handlers:

# Add these event handlers to main.py
from contextlib import asynccontextmanager
from database import database, create_tables

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifecycle manager for startup and shutdown tasks."""
    # Startup: create tables and connect
    await create_tables()
    await database.connect()
    yield
    # Shutdown: disconnect
    await database.disconnect()

app = FastAPI(lifespan=lifespan)
# ... rest of route definitions

Step 5: Migrating Error Handling Patterns

Flask's error handling with abort() and @app.errorhandler translates to FastAPI's HTTPException and custom exception handlers. Here's how to register a global exception handler for unhandled errors:

# Add to main.py
from fastapi.responses import JSONResponse
from fastapi import Request

class UnauthorizedError(Exception):
    """Custom exception for business logic errors."""
    def __init__(self, message: str):
        self.message = message

@app.exception_handler(UnauthorizedError)
async def unauthorized_handler(request: Request, exc: UnauthorizedError):
    return JSONResponse(
        status_code=status.HTTP_401_UNAUTHORIZED,
        content={"detail": exc.message, "path": request.url.path}
    )

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Catch-all handler for unexpected errors."""
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal server error", "path": request.url.path}
    )

For validation errors, FastAPI handles them automatically. When a Pydantic model fails validation, the framework returns a detailed 422 response with precise field-level error information—no extra code required.

Step 6: Migrating Middleware

Flask uses @app.before_request and @app.after_request hooks. FastAPI uses ASGI middleware with the @app.middleware() decorator. Here's a timing middleware equivalent:

# Flask middleware pattern:
# @app.before_request
# def start_timer():
#     g.start_time = time.time()
#
# @app.after_request
# def log_response_time(response):
#     elapsed = time.time() - g.start_time
#     print(f"Request took {elapsed:.3f}s")
#     return response

# FastAPI equivalent:
from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """Middleware that logs request processing time."""
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    print(f"Request to {request.url.path} took {process_time:.3f}s")
    return response

Step 7: Updating Your Test Suite

Flask tests use the app.test_client(). FastAPI provides TestClient from Starlette, which works similarly but supports async context managers. Here's how to update tests:

# tests/test_api.py
from fastapi.testclient import TestClient
from main import app
import pytest

client = TestClient(app)

def test_get_all_tasks_empty():
    """Test GET /tasks returns an empty list initially."""
    response = client.get("/tasks")
    assert response.status_code == 200
    assert response.json() == []

def test_create_task():
    """Test POST /tasks creates a new task."""
    response = client.post(
        "/tasks",
        json={"title": "Learn FastAPI", "completed": False}
    )
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Learn FastAPI"
    assert data["completed"] == False
    assert "id" in data

def test_get_single_task():
    """Test GET /tasks/{id} retrieves a specific task."""
    create_resp = client.post("/tasks", json={"title": "Test Task"})
    task_id = create_resp.json()["id"]
    response = client.get(f"/tasks/{task_id}")
    assert response.status_code == 200
    assert response.json()["title"] == "Test Task"

def test_update_task():
    """Test PUT /tasks/{id} updates a task."""
    create_resp = client.post("/tasks", json={"title": "Original Title"})
    task_id = create_resp.json()["id"]
    response = client.put(
        f"/tasks/{task_id}",
        json={"title": "Updated Title", "completed": True}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["title"] == "Updated Title"
    assert data["completed"] == True

def test_delete_task():
    """Test DELETE /tasks/{id} removes a task."""
    create_resp = client.post("/tasks", json={"title": "Delete Me"})
    task_id = create_resp.json()["id"]
    response = client.delete(f"/tasks/{task_id}")
    assert response.status_code == 204
    # Verify deletion
    get_resp = client.get(f"/tasks/{task_id}")
    assert get_resp.status_code == 404

def test_validation_error():
    """Test that missing required fields return 422."""
    response = client.post("/tasks", json={"completed": True})
    assert response.status_code == 422
    assert "detail" in response.json()

def test_not_found():
    """Test that nonexistent resources return 404."""
    response = client.get("/tasks/9999")
    assert response.status_code == 404

Run the tests with pytest tests/. The TestClient works synchronously, so you don't need async test functions for most cases.

Step 8: Running the FastAPI Application

The final step is replacing Flask's development server with Uvicorn. Update your entry point:

# Run with:
# uvicorn main:app --reload --host 0.0.0.0 --port 8000
#
# Or programmatically:
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )

Once running, visit http://localhost:8000/docs for the interactive Swagger UI and http://localhost:8000/redoc for ReDoc documentation—both auto-generated from your code.

Complete FastAPI Application Code

For reference, here is the entire migrated application assembled into a single view. Save this as main.py to run the complete example:

# Complete main.py for the migrated FastAPI application
from fastapi import FastAPI, HTTPException, status, Depends, Request
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
from typing import List
from pydantic import BaseModel, Field
from typing import Optional
import databases
import sqlalchemy
import uvicorn

# ---- Pydantic Models ----
class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    completed: bool = Field(False)

class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=100)
    completed: Optional[bool] = None

class TaskResponse(BaseModel):
    id: int
    title: str
    completed: bool
    class Config:
        from_attributes = True

# ---- Database Setup ----
DATABASE_URL = "sqlite:///tasks.db"
metadata = sqlalchemy.MetaData()
tasks_table = sqlalchemy.Table(
    "tasks", metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True),
    sqlalchemy.Column("title", sqlalchemy.String(100), nullable=False),
    sqlalchemy.Column("completed", sqlalchemy.Boolean, default=False),
)
database = databases.Database(DATABASE_URL)

async def create_tables():
    engine = sqlalchemy.create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
    metadata.create_all(engine)
    engine.dispose()

# ---- Dependency ----
async def get_db():
    return database

# ---- Application Lifespan ----
@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_tables()
    await database.connect()
    yield
    await database.disconnect()

# ---- FastAPI App ----
app = FastAPI(
    title="Task Management API",
    description="Migrated from Flask to FastAPI — Complete Example",
    version="2.0.0",
    lifespan=lifespan
)

# ---- Middleware ----
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    import time
    start = time.time()
    response = await call_next(request)
    elapsed = time.time() - start
    response.headers["X-Process-Time"] = f"{elapsed:.4f}s"
    return response

# ---- Exception Handlers ----
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal server error"}
    )

# ---- Routes ----
@app.get("/tasks", response_model=List[TaskResponse])
async def get_tasks(db=Depends(get_db)):
    tasks = await db.fetch_all("SELECT id, title, completed FROM tasks")
    return [dict(t) for t in tasks]

@app.post("/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(task_create: TaskCreate, db=Depends(get_db)):
    query = "INSERT INTO tasks (title, completed) VALUES (:title, :completed)"
    task_id = await db.execute(query, values={"title": task_create.title, "completed": task_create.completed})
    created = await db.fetch_one("SELECT id, title, completed FROM tasks WHERE id = :id", {"id": task_id})
    return dict(created)

@app.get("/tasks/{task_id}", response_model=TaskResponse)
async def get_task(task_id: int, db=Depends(get_db)):
    task = await db.fetch_one("SELECT id, title, completed FROM tasks WHERE id = :id", {"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return dict(task)

@app.put("/tasks/{task_id}", response_model=TaskResponse)
async def update_task(task_id: int, task_update: TaskUpdate, db=Depends(get_db)):
    task = await db.fetch_one("SELECT id, title, completed FROM tasks WHERE id = :id", {"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    update_values = task_update.model_dump(exclude_unset=True)
    if update_values:
        set_clause = ", ".join(f"{key} = :{key}" for key in update_values)
        await db.execute(f"UPDATE tasks SET {set_clause} WHERE id = :id", {**update_values, "id": task_id})
    updated = await db.fetch_one("SELECT id, title, completed FROM tasks WHERE id = :id", {"id": task_id})
    return dict(updated)

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int, db=Depends(get_db)):
    task = await db.fetch_one("SELECT id FROM tasks WHERE id = :id", {"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    await db.execute("DELETE FROM tasks WHERE id = :id", {"id": task_id})

if __name__ == "__main__":
    uvicorn.run("__main__:app", host="0.0.0.0", port=8000, reload=True)

Best Practices for a Smooth Migration

Handling Common Migration Challenges

Challenge: Large File Uploads

Flask handles file uploads via request.files. In FastAPI, use UploadFile:

from fastapi import File, UploadFile

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    # Process file contents
    return {"filename": file.filename, "size": len(contents)}

Challenge: Background Tasks

Flask uses Celery or @app.after_request for deferred work. FastAPI has built-in BackgroundTasks:

from fastapi import BackgroundTasks

@app.post("/send-notification")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_async, email)
    return {"message": "Notification will be sent"}

Challenge: CORS Configuration

Flask uses flask-cors. FastAPI uses CORSMiddleware:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_methods=["*"],
    allow_headers=["*"],
)

Challenge: Flask Extensions Without FastAPI Equivalents

For Flask extensions like Flask-Login or Flask-Admin, you have options: wrap the Flask app with WSGIMiddleware for gradual migration, find ASGI-native alternatives (e.g., FastAPI-Users for authentication), or implement the functionality directly using FastAPI's dependency injection and middleware.

Conclusion

Migrating from Flask to FastAPI is a strategic investment that pays dividends in performance, developer productivity, and API quality. The process follows a clear pattern: replace implicit Flask conventions with explicit FastAPI declarations, convert synchronous route handlers to async where beneficial, introduce Pydantic models for robust data validation, and restructure error handling and middleware using FastAPI's modern patterns. By following this step-by-step guide and embracing incremental migration, you can transform a legacy Flask application into a high-performance, self-documenting FastAPI service without disrupting existing functionality. The auto-generated interactive documentation alone often justifies the migration effort, providing your API consumers with an always-accurate, explorable interface that Flask could only achieve with additional, manually-maintained plugins. Start with a single endpoint, validate with tests, and build momentum from there—the path from Flask to FastAPI is well-paved and thoroughly rewarding.

🚀 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