Migrating from Flask to FastAPI: What Actually Went Wrong

Most "Flask to FastAPI" articles show you a hello-world side by side and call it a day. This one is about the other kind of migration: a production service with a REST API, a WebSocket endpoint, background workers, three environments, and callers you can't ask to change anything.

The service is an internal tool: callers create jobs through REST, workers poll a log backend, and results stream back over a WebSocket. It ran on Flask-RESTful behind gunicorn, plus Flask-uWSGI-WebSocket behind uWSGI. We moved all of it to FastAPI on uvicorn without a single change on the caller side.

What You'll Learn

✅ Why we split the migration into phases and what each phase shipped
✅ How to keep the API contract byte-for-byte identical (including error bodies)
✅ Why we kept the handlers synchronous, at least at first
✅ WebSocket pitfalls that only show up once you leave uWSGI
✅ How we proved there was no regression

1. The Real Reason to Migrate

It wasn't speed. Flask was never our bottleneck (we'll get to what was). The reasons were maintenance:

• A pile of Flask-only extensions: Flask-RESTful, flask-restful-swagger-2, Flask-SQLAlchemy, Flask-Migrate, Flask-Babel, Flask-Cors, Flask-uWSGI-WebSocket. Each is a separate release cycle and a separate way to break on a Python upgrade.
• WebSocket on uWSGI is a niche setup. Few people know it, and it blocks upgrading Python.
• Hand-written Swagger 2 that drifted from the code. FastAPI generates OpenAPI from the routes.

When we dropped the dependencies, requirements.txt lost eleven direct packages (Flask, Flask-RESTful, Flask-SQLAlchemy, uWSGI, gunicorn and friends). Fewer moving parts was the whole win.

2. Phase 0: Find the Real Bottleneck First

Before touching the framework we profiled. MySQL was at 93% CPU. The jobs table had only a primary key, and the job loop ran WHERE redis_job_id = ? AND status = 1 for every active job, once per second. Each query scanned ~175k rows.

Three composite indexes later, the hot query went from ~95 ms to under 1 ms.

A framework migration would have changed nothing about this problem.

If you're migrating "for performance", profile first. We would have blamed Flask for a missing index.

3. The Phased Plan

We never had a "big bang" branch. Each step was deployable on its own and left the service working:

Step A is the one people skip, and it's the most valuable. Workers, the worker manager and the CLI all used current_app and app_context(). Until those ran without a Flask app, nothing else could move. Step A also fixed a circular import that only worked because the worker created a Flask app first.

# app/core.py: one entry point for web, workers, and CLI
def init(env):
    settings = load_settings(env)
    setup_logging(settings)
    database.init(settings)
    redis_provider.init(settings)
    return settings

4. Keep the Contract Identical

Callers don't care what framework you use. They care that 400 {"message": {"name": "..."}} stays exactly that.

FastAPI's defaults differ from Flask-RESTful in ways that break clients:

• Validation errors return 422 with a detail list. We had 400 with message.
• Errors use the key detail. We used message.
• Flask-RESTful's reqparse accepted values from the JSON body, form, and the query string.

Idiomatic FastAPI (Pydantic models for every body) would have changed the behavior. So for a migration we deliberately did the un-idiomatic thing and wrote a tiny reqparse-compatible parser:

class ApiError(Exception):
    def __init__(self, status_code, body):
        self.status_code, self.body = status_code, body
 
async def api_error_handler(request, exc):
    return JSONResponse(status_code=exc.status_code, content=exc.body)
 
api.add_exception_handler(ApiError, api_error_handler)

def parse_args(values, arguments):
    # first error wins, in declared order, like reqparse
    return {a.name: a.parse(values) for a in arguments}

It's 60 lines. Idiomatic Pydantic models can come later, as a separate change, once the migration is behind you and clients can be versioned.

The same logic applies to paths. We mounted the API as a sub-app at /api/v1, so the OpenAPI paths stay relative and every URL stays the same.

api = FastAPI(title="My API", openapi_url="/swagger.json", docs_url="/docs/")
app = FastAPI(openapi_url=None, docs_url=None)
app.mount("/api/v1", api)

5. Sync Handlers Are Fine

The most common advice is "use async def everywhere". We didn't, and I'd do it again.

Our database layer is synchronous SQLAlchemy with mysqlclient. A plain def handler runs in FastAPI's thread pool, which is exactly what we need. Writing async def and calling blocking code inside it would have been worse than Flask, because it blocks the event loop for everyone.

@router.post("/jobs")
def create_job(request: Request, ...):   # plain def: runs in a worker thread
    ...

One catch: SQLAlchemy's scoped session is bound to the thread, and FastAPI's pool reuses threads. A session left open by one request leaks into the next, and holds a transaction (and a metadata lock) forever. We wrap every handler:

def db_session(handler):
    @functools.wraps(handler)
    def wrapper(*args, **kwargs):
        try:
            return handler(*args, **kwargs)
        finally:
            db.session.remove()   # always end the session in its own thread
    return wrapper

The same wrapper later grew a retry on MySQL deadlocks. Flask-SQLAlchemy used to do the remove() for you on app-context teardown, so you don't notice it until it's gone.

6. WebSocket: Where the Surprises Were

Leaving uWSGI for uvicorn made the WebSocket code simpler, but it exposed things the old stack had hidden.

A disconnected client kept consuming messages. The old handler kept popping the caller's messages from Redis after the socket closed, and those messages were lost. In an async handler, WebSocketDisconnect ends the loop at once, and a message that can't be sent goes back to the queue. This was a known bug that the rewrite fixed for free.

Blocking calls in the stream. The streamer is async, but Redis and database calls are blocking. They run in worker threads:

message = await asyncio.to_thread(redis_pop, caller_id)

One Redis connection per poll. The old code created a new connection every 0.1 s. We moved to a shared pool.

An open transaction for the lifetime of the socket. The handler authenticated the API key with a database query and never ended the transaction. With callers connected for ~30 minutes, that transaction held a metadata lock, and the next ALTER TABLE migration would have timed out. We found it on the integration environment, not in tests. The fix was one line (db.session.remove() right after the auth check), plus a test that asserts the handler starts streaming with no open transaction.

If your WebSocket handler touches the database, check what it is still holding after the handshake.

7. Deployment: The One Minute of 502s

The reverse proxy changed from uwsgi_pass to proxy_pass. Because the config is applied before the rolling restart, the old uWSGI container briefly received plain HTTP. We got about a minute of WebSocket 502s, once. Messages wait in Redis, so nothing was lost, but we wrote it down in the commit message so nobody was surprised. Plan for this kind of one-time glitch instead of hoping it won't happen.

8. How We Knew It Worked

The key decision: the integration tests never import the app. They talk HTTP and WebSocket to a running stack, so the same suite ran before and after the rewrite and checked what callers actually see.

• 64 black-box tests against the real stack (REST contract, WebSocket, job → worker → log backend → WebSocket)
• 150+ in-process unit tests
• A load test with the same scenarios before and after

No regression, with slightly tighter tails. That's an honest result: FastAPI didn't make this service faster, because the framework was never what limited it. It made the service maintainable.

Summary

✅ Profile before you blame the framework. Ours was a missing index.
✅ Decouple your core (DB, config, workers) from the web framework first.
✅ Migrate in deployable steps, REST then WebSocket, never a big bang.
✅ Keep the contract: error shapes, status codes, and argument sources matter more than idioms.
✅ Sync handlers on a sync database are fine; don't write async def around blocking calls.
✅ Always end the DB session in the thread that opened it, and check long-lived WebSocket handlers for open transactions.
✅ Test black-box so the same suite judges the old and new implementation.

If you're deciding between threads and coroutines inside the migrated service, read Migrating from ThreadPool to AsyncIO in Python, and for the FastAPI fundamentals see Asynchronous Programming in FastAPI.

Related Posts

• From RQ Workers to an Asyncio Engine
• Upgrading Django 5.2 to 6.0
• Upgrading Docker and Compose: The Forgotten Cron Job