Skip to content

Migrating from Flask to FastAPI, Part 3

This is the third and last part of these blog series.

The previous parts are:

Here you will see the advanced techniques to handle apps with complex patterns that are deeply integrated with Flask's internals. And also additional tips and tricks to do the actual migration work, how to guide yourself in the code base, etc.

Request Local State

This is probably the most important and, in some cases, the most difficult part of migrating from Flask to FastAPI. How difficult or complex it is will depend mostly on how you structured your code base.

I'll show you the tips to handle and migrate what's necessary.

Flask Request State

In Flask, because of its design, there are some pseudo-global objects that have data that in reality only applies to the specific request being handled.

The main two objects you would see are:

  • request: holds the data from the current request being handled.
  • g: this is a global wildcard object, you can set attributes in it and retrieve them in other places in the codebase. And this will only apply to the current request.

Because these objects are pseudo-globals, there's a chance that you are using them at some point inside in your internal functions that handle internal steps.

import secrets

from flask import Flask, abort, g, jsonify, request

internal_token = "thisshouldbeanenvironmentvariable"

app = Flask(__name__)


def process_data():
    data = request.get_json()
    item_name = data["item_name"]
    token = g.token
    print(f"Do something with item: {item_name} using token: {token}")


@app.before_request
def verify_token():
    token = request.headers.get("x-token", "")
    if not secrets.compare_digest(
        token.encode("utf-8"), internal_token.encode("utf-8")
    ):
        abort(401)
    g.token = token


@app.route("/items/", methods=["POST"])
def create_item():
    data = request.get_json()
    token = g.token
    process_data()
    return jsonify({"data": data, "token": token})

In this example, process_data() doesn't receive parameters, but extracts data from the Flask's request and the g object.

FastAPI Request State

In FastAPI there are no pseudo-global objects, any data extracted or created would be defined somehwere in function parameters with type annotations.

So, in FastAPI, you would probably pass any data needed directly via function parameters down the line.

import secrets

from fastapi import FastAPI, Header, HTTPException, Depends
from pydantic import BaseModel

internal_token = "thisshouldbeanenvironmentvariable"

app = FastAPI()


class Item(BaseModel):
    name: str
    size: int = 0
    tags: set[str] = set()


def process_data(data: Item, token: str):
    item_name = data.name
    print(f"Do something with item: {item_name} using token: {token}")


def verify_token(x_token: str = Header()):
    if not secrets.compare_digest(
        x_token.encode("utf-8"), internal_token.encode("utf-8")
    ):
        raise HTTPException(401)
    return x_token


@app.post("/items/")
def create_item(item: Item, token: str = Depends(verify_token)):
    process_data(data=item, token=token)
    return {"data": item, "token": token}

In the FastAPI example, the process_data() function would receive the data it needs as parameters.

Here data would be an instance of the Pydantic model, but you could also refactor it to be a plain dictionary if you wanted.

The good news is that although FastAPI depends on and uses Pydantic, Pydantic itself doesn't depend on FastAPI. So, you could actually use the same Pydantic models from your FastAPI app to manually validate and serialize Flask data, and then use them down the line in your internal code logic.

State, Pseudo-Globals, and Compatibility

The pseudo-global objects from Flask, like request and g, work for Flask applications, but there are a couple of caveats.

If you have a function like the first process_data():

# More code here 👈

def process_data():
    data = request.get_json()
    item_name = data["item_name"]
    token = g.token
    print(f"Do something with item: {item_name} using token: {token}")

# More code here 👈

...and try to use it outside of a Flask application, for example in a background job (e.g. Celery, RQ), it won't work, because it depends on these pseudo-global objects that are only available and with data when used in a Flask application, during a request.

The same would happen if you tried to use the same function shared by a Flask application and a FastAPI application (or anything else). It would only work inside of the Flask application.

If you have this type of pattern in your code base, this would be the main thing you have to pay special attention to during your migration.

Refactor Flask App Data Handling

Let's imagine the most difficult scenario, you want to keep both a Flask app and a FastAPI app, both at the same time, and sharing some internals.

In this case, something you could do, would be to, ideally, only use the Flask-specific objects inside of Flask-specific code, so, it would be in the Flask route function.

Then ideally you would pass any needed data as function parameters.

Refactor Data: Models

You could have a models.py module (file) with Pydantic models, it could have the data models for FastAPI, for example:

models.py
from pydantic import BaseModel


class Item(BaseModel):
    name: str
    size: int = 0
    tags: set[str] = set()

Because Pydantic is independent of FastAPI you could take advantage of the same models in other places, even (manually) in Flask.

Refactor Data: Utils

Now let's say that the additional functions that handle the data, independent of the framework, could be put in a file uitls.py, this file could import the same Pydantic model we defined for the type annotation, and do the work that only deals with data, but doesn't depend on any web framework:

utils.py
from .models import Item


def process_data(data: Item, token: str):
    item_name = data.name
    print(f"Do something with item: {item_name} using token: {token}")

Refactor Data: FastAPI

Then, in the FastAPI app, you would use the same Pydantic model and utility function, but you would import them from those shared modules:

fastapi_app.py
import secrets

from fastapi import FastAPI, Header, HTTPException, Depends

from .models import Item
from .utils import process_data

internal_token = "thisshouldbeanenvironmentvariable"

app = FastAPI()


def verify_token(x_token: str = Header()):
    if not secrets.compare_digest(
        x_token.encode("utf-8"), internal_token.encode("utf-8")
    ):
        raise HTTPException(401)
    return x_token


@app.post("/items/")
def create_item(item: Item, token: str = Depends(verify_token)):
    process_data(data=item, token=token)
    return {"data": item, "token": token}

Refactor Data: Flask

And here's probably the most interesting part, in Flask, you would extract the information from the request and g object in the Flask route function that is purely Flask-dependent.

Then you would just pass that pure data down the line. And you can re-use the same Pydantic models to take advantage of the data validation, autocompletion, inline errors in your editor, etc.

flask_app.py
import secrets

from flask import Flask, abort, g, jsonify, request

from .models import Item
from .utils import process_data

internal_token = "thisshouldbeanenvironmentvariable"

app = Flask(__name__)


@app.before_request
def verify_token():
    token = request.headers.get("x-token", "")
    if not secrets.compare_digest(
        token.encode("utf-8"), internal_token.encode("utf-8")
    ):
        abort(401)
    g.token = token


@app.route("/items/", methods=["POST"])
def create_item():
    raw_data = request.get_json()
    data = Item.parse_obj(raw_data)
    token = g.token
    process_data(data=data, token=token)
    return jsonify({"data": data, "token": token})

This process would probably be the ideal. There might be cases where one function calls another, and that calls another, and that another, and deep down the call chain is the place where the data from the pseudo-global objects is used.

If you can, still, try to refactor the functions to pass the data through parameters.

That will make all your code much more explicit aobut what it is doing, more obvious, and a lot easier to debug.

...but, if the call chain is really too complex and you just can't refactor all the call chains, you can do some black magic to handle it.

Pseudo-globals with Context Variables

Warning

If you can do the above refactor, that would be recommended.

In that case, you can skip this section, there's quite some black magic here that you probably don't need, and you can skip to the next section.

Now, let's say that you have a long function call chain, and deep down that is that you are using that request or g pseudo-global objects.

And it's just unfeasible to refactor all that.

Then you can use another tool to create pseudo-global objects that would be compatible with async code (FastAPI is an async framework underneath).

You might have heard of "thread-locals" or "thread-local variables". That refers to pseudo-global variables that look in the code as if they were globals, but that in reality have an independent value per thread.

That has been the main advanced trick to handle these situations.

But thread-locals depend on the asumption that any code that is doing work concurrently (during the same period of time) as other code, would be running in its own thread. Each concurrent execution of code would be running in its own thread.

...but now we have async and await, which improve a lot how we think about concurrency and how we work with it, including performance benefits, etc.

And having async and await means that now we can have concurrency (multiple executions of code during the same period of time) that happen on the same thread. Because the async event loop runs on a single thread.

This means that now we can't simply talk about "threads" and consider them the only possible separation for concurrency, and a thread-local variable would not work in this case, as its value depends just on the thread, but there might be several executions of async code running concurrently on the same thread, using the same thread-local variable.

So, we need something that is not just thread-safe, but concurrent-safe, that can also handle async "contexts". An async context is just that flow of execution of an async task/function call.

And in FastAPI, and other use cases, you can also call blocking code from an async function by sending it to a worker thread, but that's still work that belongs to this same async task/context. If it was using thread-locals, that would mean that work that belongs to the same async context/task, but is running on another thread because it's blocking, would not have the same value of the thread-local as it should.

Fortunately, modern versions of Python (the ones currently supported, Python 3.7 and above) have an alternative to thread-locals that is compatible with async contexts (and also compatible with threads), it's called Context Variables, they are in the contextvars module in the standard library in Python.

Almost anything that used thread-local variables, should now use context variables to be compatible with modern code, including async and await.

...are you still reading? Wow, awesome! Let's see some code then.

Flask: Deeply Nested Pseudo-Globals

Let's say that your code has some functions that probably process data in some way, for example, your Flask code looks like this:

flask_app.py
import secrets

from flask import Flask, abort, g, jsonify, request

from .utils import do_work4

internal_token = "thisshouldbeanenvironmentvariable"

app = Flask(__name__)


@app.before_request
def verify_token():
    token = request.headers.get("x-token", "")
    if not secrets.compare_digest(
        token.encode("utf-8"), internal_token.encode("utf-8")
    ):
        abort(401)
    g.token = token


@app.route("/items/", methods=["POST"])
def create_item():
    data = request.get_json()
    token = g.token
    do_work4()
    return jsonify({"data": data, "token": token})

And then those functions call each other, and deep down the call stack, at some point, one of those functions depends on Flask's pseudo-global variables, like g and request.

utils.py
from flask import g, request


def process_data():
    data = request.get_json()
    item_name = data["item_name"]
    token = g.token
    print(f"Do something with item: {item_name} using token: {token}")


def do_work1():
    return process_data()


def do_work2():
    return do_work1()


def do_work3():
    return do_work2()


def do_work4():
    return do_work3()

Refactoring all those calls to pass the data directly might seem just too much work.

And possibly some of these functions are also called in other places, and you would have to refactor all those places too. The refactor could seem so costly that you would just consider missing on the FastAPI advantages and not migrating, just because refactoring all that code would be too costly.

So, I'll teach you how you can use context variables to solve it. 🚀

Tip

Even if you do this, I would encourage you to plan on refactoring that later to pass the data directly.

It would simplify your code a lot and would make it a lot easier to debug and understand future problems.

Flask: Deeply Nested Context Variables

Again, we will assume the most complex scenario. That you need to keep the code compatible with both the Flask app and the FastAPI app.

What we will do is replace the Flask pseudo-globals with context variables. We'll use these context variables in the utility functions and set their values at the Flask level, pretty much the same way as we did with Flask's pseudo-global objects.

This way, later we can also set the same context variables on the FastAPI side, and use the same utility functions that consume those context variables.

So, let's refactor these utils to use context variables:

utils2.py
from contextvars import ContextVar
from typing import Any

request_data = ContextVar[dict[str, Any]]("request_data")
request_token = ContextVar[str]("request_token")


def process_data():
    data = request_data.get()
    item_name = data["item_name"]
    token = request_token.get()
    print(f"Do something with item: {item_name} using token: {token}")


def do_work1():
    return process_data()


def do_work2():
    return do_work1()


def do_work3():
    return do_work2()


def do_work4():
    return do_work3()

Context variables have several requirements and details:

  • They have to be declared at the top level of the module (i.e. outside of functions).
  • The context variable needs to have as the "name" (the string passed as the first parameter) the same name of the variable.
  • You access the value with the get() method, as in some_variable.get().

You can declare the type of the value using the square brackets. This way, when you access the context variable in the code, your editor will be able to give you autocompletion, inline errors, etc.

Now, let's refactor the Flask app to set the values for these context variables:

flask_app2.py
import secrets

from flask import Flask, abort, jsonify, request

from .utils2 import do_work4, request_token, request_data

internal_token = "thisshouldbeanenvironmentvariable"

app = Flask(__name__)


@app.before_request
def verify_token():
    token = request.headers.get("x-token", "")
    if not secrets.compare_digest(
        token.encode("utf-8"), internal_token.encode("utf-8")
    ):
        abort(401)
    request_token.set(token)
    data = request.get_json()
    request_data.set(data)


@app.route("/items/", methods=["POST"])
def create_item():
    data = request_data.get()
    token = request_token.get()
    do_work4()
    return jsonify({"data": data, "token": token})

To set the value of a context variable you use the set() method, as in some_variable.set("foo").

Depending on how you deploy and start your Flask application, the context variable values could reset automatically for each request, which is what you would want. But in case the way you deploy it doesn't do that, you would probably want to make sure you always set the value at the beginning of the request. In this case, we do it in the @app.before_request.

The same way that you access context variables in the utility functions, you can now access them in the Flask routes. You could still use the same g and request object there, but to make the code more consistent and avoid multiple app states of running your code, you might want to use the same variables everywhere.

FastAPI: Deeply Nested Context Variables

Now, if you check again the file utils.py above, you will see that it no longer imports anything from flask. Now those utility functions are independent of framework.

And now we can add the wiring to set those context variables in FastAPI.

fastapi_app.py
import secrets

from fastapi import Depends, FastAPI, Header, HTTPException, Request

from .models import Item
from .utils2 import do_work4, request_data, request_token

internal_token = "thisshouldbeanenvironmentvariable"

app = FastAPI()


async def set_global_request_token(x_token: str = Header()):
    request_token.set(x_token)  # (1)!
    return x_token


def verify_token(token: str = Depends(set_global_request_token)):  # (2)!
    if not secrets.compare_digest(
        token.encode("utf-8"), internal_token.encode("utf-8")
    ):
        raise HTTPException(401)
    return token


async def set_global_request_data(request: Request):  # (3)!
    data = await request.json()
    request_data.set(data)


@app.post("/items/", dependencies=[Depends(set_global_request_data)])  # (4)!
def create_item(item: Item, token: str = Depends(verify_token)):
    do_work4()
    return {"data": item, "token": token}

  1. We are setting the context variables in dependency functions. For the token, we just extract it and set the context variable, and we return it again.

  2. Then we use that dependency set_global_request_token() as a sub-dependency in the dependency function verify_token().

  3. For the request data, we have a dependency set_global_request_data(). This one doesn't even return anything. We just want it to get the request data as JSON and set it in the context variable.

  4. Because the dependency set_global_request_data() doesn't return anything, we don't need its value, we don't have to put it in the parameters of the path operation function create_item(), that would mean adding a dummy parameter that is not used.

    That could be confusing for other developers, and you could have tools and editors suggesting to remove it because you are not using that parameter.

    So, you can pass it in the argument dependencies=[] to the path operation decorator.

There are several things going on to notice here. Click the bubbles to see what's happening in each code section.

Probably the most important and subtle detail, these dependency functions that set the context variables have to be async. And we don't want to run potentially blocking code in async functions (let's imagine that verifying the token was expensive), so we can separate the logic to set the context variable in an async function and leave the rest in the blocking function.

Technical Details: ContextVars in async Functions

If you want to understand the very technical details of why we have to put the context variables in an async dependency, keep reading. Otherwise, you can probably just do it and continue with your code.

Here's the thing, when you declare a path operation function in FastAPI with async def, it will be run normally in the async event loop in the main thread.

If you declare the function without async def, just using regular def, then FastAPI asumes that you could have blocking code inside (as would probably be the case if you come from Flask). And then FastAPI (actually Starlette underneath) runs it on a thread worker, that way it doesn't block the main event loop, and you can get the best performance possible given the code. This is what allows you to start using FastAPI right away even if you don't use any async code.

Now, on top of that, if you define async def dependency functions, again, FastAPI will run them on the async event loop on the main thread. And if you define dependency functions with regular def, FastAPI will asume that they might have blocking code, and will run them on a worker thread, to not block the main async event loop.

This even means that you can mix and match async and non-async dependencies, and FastAPI will run them on the right places.

If you want to learn more about that, check the FastAPI Async docs.

Now, having that background of how FastAPI works with async and non async code, we can finally discuss why you need to set the context variables in async functions. 😅

In standard Python (independent of FastAPI), when you start an async function, and then at some point you want to call a blocking function, to not block the main event loop, you manually send that function to be run in a worker thread (the same thing FastAPI would do automatically for dependencies).

If you set context variables in the async function, that "context" (the set of context variables set with their values) is copied to the thread worker, this way the functions in the thread worker can access the same values.

🚨 But, any context variables that you set inside of a thread worker are not copied back outside of the thread worker.

Let's see what this means in an example:

  • You have an async function async def main()
  • this function async def main() sets a context variable current_user to the value "Rick"
  • then the function async def main() uses/calls a blocking function def do_work() sending it to a worker thread
  • the function accesses the value current_user, and gets the value "Rick", that was set in the "parent context" (in the async def main() function)
    • this works as expected ✅
  • but then the blocking function def do_work() that is running on a thread worker tries to set the context variable of current_user to the value "Morty"
    • this will work inside of the same function, and in internal calls to other functions ✅
    • but it won't work outside of it, outside of this thread, once async def main() continues executing, after having called def do_work(), the value of current_user will still be "Rick", instead of "Morty" 🚨

Because of that, if you used a regular def function for a FastAPI dependency that sets a context variable, it would be run on a worker thread, and the value would be lost for the rest of the execution. You want the value to be persisted for any children/related code in that request. That's why you have to set the context variable in the async dependency function.

Tip

If for some reason you really need to somehow set the value in a regular def dependency function, you can do a simple trick. You cannot set there the value of the context variable, you still have to set it in a parent async dependency function, but...

You can mutate the value of the context variable in the regular def function. So, the parent async dependency function can set the context variable to an empty dict, and then you can set a value in that dict in the blocking/regular def dependency function.

Because changing a value in a dict doesn't change the actual object in the context variable (it still refers to the same object), those changes to that dictionary will be reflected outside.

As a shortcut to think about it, you can't use some_context_var.set() inside of regular def dependency functions, but you can use some_context_var.get() however you want.

Concurrency, Gevent, Async and Await

One of the most complex sections of code in the migration was dealing with concurrency using Gevent, and migrating to async and await, while keeping it compatible with non-async code, making it work with FastAPI and Flask.

We were using Gevent with Flask, monkeypatching internals to get that magic automatic concurrency without using async and await. Doing that, without async and await, brings a whole range of difficulties when debugging code, but that's a topic better covered in other existing articles.

As we were now using FastAPI for several of the APIs we were migrating, and FastAPI is underneath an async framework (with compatibility for non-async code), it made sense to migrate away from Gevent and take advantage of async and await in the sections that required concurrency, to get that better predictability, make things easier to debug, etc.

At the same time, we needed to be able to keep using that code in some of the small components that stayed in Flask, but Flask is not async underneath, so we needed the code to be compatible with both.

That's why during a Christmas break I built Asyncer, to simplify writing those interactions of async and blocking code and doing it the right way, without blocking the async event loop on the main thread, etc.

Code with Gevent

Let's say that we have this file with some functions to process data:

utils.py
import gevent
import requests


def get_data(group_id: int, item_id: str):
    response = requests.get(f"https://example.com/{group_id}/{item_id}")
    return response.json()


def do_heavy_work(data: dict):
    for i in range(999_999_999):
        # some slow CPU-bound work here
        print("Here would be work")
    return data


def process_item(group_id: int, item_id: str):
    data = get_data(group_id=group_id, item_id=item_id)
    processed_data = do_heavy_work(data)
    return processed_data


def process_items(item_group_ids: list[tuple[int, str]]):
    greenlets = [
        gevent.spawn(process_item, group_id, item_id)
        for group_id, item_id in item_group_ids
    ]
    gevent.joinall(greenlets, raise_error=True)
    items = [greenlet.value for greenlet in greenlets]
    return items

With the Gevent concurrency handling, the function get_data() will use, at some point, deep down there (somewhere inside of requests' internal code), one of the low-level functions from Python that would normally block waiting, but with Gevent monkeypatching that, now it's not gonna block and Gevent will call other things concurrently while waiting for that.

A possible problem is that everything is still running on the same thread, and if do_heavy_work() takes a long time, consuming CPU, that's gonna block the main thread. For it to allow other code to run concurrently it would have to call Gevent/Greenlet specific code that would tell it to go and do some other work before continuing here.

You would use those functions directly in Flask like this:

flask_app.py
from gevent import monkey

monkey.patch_all()

from flask import Flask, jsonify, request

from .utils import process_items

app = Flask(__name__)


@app.route("/items/", methods=["POST"])
def handle_items():
    data = request.get_json()
    items = process_items(data)
    return jsonify({"items": items})

We want the Flask app to keep its code as is, we want to only remove the Gevent monkeypatching and change the internal implementation of process_items().

Refactor Gevent to Async and Await with Asyncer

What we want to do is not to change all the code base to use async and await everywhere, that's normally an unfeasible task, at least in one go.

But we want to be able to use async code in the right parts, and still keep using the regular blocking code everywhere else, and send those blocking calls to thread workers where necessary.

In some way, we need to be able to manually do what FastAPI normally does underneath when you use regular def and async def functions for path operations and dependencies.

In general, we have to run async functions in the async event loop on the main thread. And we want blocking functions running on a thread worker (one of several threads created automatically).

And when one of these blocking functions needs to call async code, it has to send that execution to the async event loop on the main thread.

graph TB
  A[Async event loop on main Tthread] --> B{Async function?};
  B -->|Yes| A;
  B -->|No| C[Thread worker];
  C --> B;

If all our code was compatible with async and await, all that logic would look and feel a lot simpler. But as we need to be able to mix both async and blocking code, and we need the code to go from the async main thread to the worker thread, and from that same function running in the worker thread we might need to execute async code, and so on, it can be complex to wrap your head around it.

The important thing is to keep in mind that the code will only be executed in one of two levels, the async event loop on the main thread or in a thread worker.

Here's how we can refactor the concurrent code using Asyncer:

utils2.py
import asyncer
import requests
from asyncer import asyncify, syncify


def get_data(group_id: int, item_id: str):
    response = requests.get(f"https://example.com/{group_id}/{item_id}")
    return response.json()


def do_heavy_work(data: dict):
    for i in range(999_999_999):
        # some slow CPU-bound work here
        print("Here would be work")
    return data


def process_item(group_id: int, item_id: str):
    data = get_data(group_id=group_id, item_id=item_id)
    processed_data = do_heavy_work(data)
    return processed_data


async def async_process_items(item_group_ids: list[tuple[int, str]]):  # (1)!
    async_process_item = asyncify(process_item)  # (2)!
    async with asyncer.create_task_group() as tg:  # (3)!
        soon_values = [
            tg.soonify(async_process_item)(group_id=group_id, item_id=item_id)  # (4)!
            for group_id, item_id in item_group_ids
        ]
    items = [soon_value.value for soon_value in soon_values]  # (5)!
    return items


def process_items(item_group_ids: list[tuple[int, str]]):
    return syncify(async_process_items, raise_sync_error=False)(  # (6)!
        item_group_ids=item_group_ids
    )

  1. First, we move the logic that will create concurrent calls to an async function async_process_items() (instead of the regular process_items()).

  2. We want to call process_item(), that is a blocking function, so we want to execute that in a thread worker instead of the async event loop on the main thread.

    So, we use asyncify() to create a version of the same function that will be executed on a thread worker and that we can await. It's like an async version of that same function.

  3. Next, we create an Asyncer task group (which is just an AnyIO task group with a bit of extra niceties to handle types better).

    Having a task group is what allows us to run multiple tasks (async functions) concurrently, and gives us the guarantee that the async code will be finished running by the point the async with block ends.

    It would also allow us to catch all the possible exceptions that could raise from those async functions in an ordered way, but we won't do that here to keep the example simple.

  4. Then we use the task group's method soonify() to tell the Asyncer task group that we want it to concurrently call the function async_process_item() with those parameters.

    Calling that soonify() will return a special SoonValue object that will contain the value returned after we finish the async with block.

  5. Then we can extract al those values from the SoonValue objects and return them.

  6. Because this function async_process_items() is async, but we need to call it from blocking code, we can use syncify() to call it on the async event loop in the main thread... that's what syncify() would normally do, but in the case of Flask, there's no async event loop, the blocking code is already running on the main thread.

    So, running plain syncify() there would raise an error about that. We can set the parameter raise_sync_error=False to tell it to check if there's an async event loop, and if there's none, instead of raising an error, create one right there just to execute this.

    This wouldn't have as good performance as if there was an existing async event loop, but it will allow us to call all this async code inside of Flask. This is also the same trick that new versions of Flask do to allow defining async route handlers, even though underneath it is not an async framework.

Now with all the pieces in place, we can use this new process_items() function from Flask, the only thing we change is remove the Gevent monkyepatching.

flask_app2.py
from flask import Flask, jsonify, request

from .utils2 import process_items

app = Flask(__name__)


@app.route("/items/", methods=["POST"])
def handle_items():
    data = request.get_json()
    items = process_items(data)
    return jsonify({"items": items})

For the FastAPI app, we can also just call the function directly. We could even define this path operation to be an async function and then call the async version async_process_items(), but for this example, to show that we can now use the same function that handles concurrent calls in both Flask and FastAPI, we'll leave it as a regular function that calls the same process_items():

fastapi_app.py
from fastapi import FastAPI

from .utils2 import process_items

app = FastAPI()


@app.post("/items/")
def handle_items(data: list[tuple[int, str]]):
    items = process_items(data)
    return {"items": items}

Refactoring Tips

Now that you know all the technical details of the code and how to handle it, how to structure it, what and how to refactor, the "only" step missing is going and doing it in your code base.

This can be a daunting and confusing task in several ways, you need to somehow review your code base in an ordered way, checking all that matters, refactoring what could be needed, and not getting lost in the middle. Here are some simple tips for doing that work of refactoring the code.

Explore Code and Comments

You normally start in one function, reading the definition, it calls another function and you have to go and check the definition of that function, then from there, that one calls other two functions and you have to go and check the definition for those...

And you might also end up reviewing the same function multiple times, etc.

At some point, it's quite possible that you would forget where in the code you were, what was the next step, what was the next line to check.

There's a couple of simple tricks you can do. You can leave some breadcrumbs in your code that will help you get back to the point where you were at.

One option is code comments, like:

# TODO (flask-refactor): continue here
def some_function():
    # do stuff

Then you could use extensions in your editor to show you the TODOs in the code base. And some of those extensions and tools might have a way to filter by text, so you could mark the comment with some keyword, your username, etc. That way you can filter only the TODO comments that are relevant to you.

There are other extensions that allow you to create bookmarks, even without comments and without changing the code. You can use anything that works for you.

The trick is, right before clicking the "Go to definition" or "Go to reference" option in your editor that will take you somewhere in the code, create one of this breadcrumbs, so that once you finish checking that function, you can come back to this one.

Having some type of breadcrumbs is very useful in particular when you need to start in a function, then in the middle of that, before finishing reading this function, go to the definition of another function that is called there, and from that one, to another... you end up traversing the tree of function calls going first deep down through the calls, instead of reading the entire definition body of the first function before starting to read the definition body of internal functions (so, it's a "depth-first search" instead of "breadth-first search").

And because you end up "in the middle" of reading the definition body of a bunch of functions that call each other, having the breadcrumbs can help you track your steps back, and continue reading the definition body of the last function you were checking.

Comment Migrated

The other trick you can do is, when you are done refactoring or checking one function, and you know that it's done, you can add a comment on top, for example # Migrated.

This way, whenever you are reviewing another function somehwere else in the code, let's say process_items(), and you see that it is calling another function, let's say extract_token(), when you go to the definition for extract_token(), if you see a comment # Migrated, you can avoid checking that one again, including any children function calls.

Otherwise, you could end up reviewing extract_token() many times, including anything internal that it calls, just to see if you need to change something, and reviewing it the first time was enough.

You end up doing some type of "memoization" of your own work, you already know that you mentally processed that function (and maybe even refactored it), you don't need to go back again there, it's already done. So, you can go back up through the breadcrumbs and continue reviewing process_items().

Go to Reference and g

If you are migrating from Flask, one of the main things you will have to do is clean any code that uses Flask's g or request objects, and as you don't pass them through arguments but any function could be importing and using them, it might not be obvious how to find where you have to change something.

When you hit Ctrl + Click (on Linux/Windows) or Cmd + Click (on Mac) on a variable/function call, etc. it normally takes you to the definition. If in the definition you hit Ctrl + Click or Cmd + Click again, it will take you to the list of references, the places that use that object in your code.

If you hit that on Flask's g object, it will take you to the definition, inside of Flask's own code. If you try to hit Ctrl + Click or Cmd + Click on g right there, it might not work, because it could take you only to the references inside of Flask's own code. Or it could try to find the references in your whole environment, in each of the packages installed.

But, if you import g in your code, and manually right-click (or find the option in some way) to find the references of that object (instead of the definition), you will probably see all the places that you need to refactor in your code.

The same would work for Gevent, or any other library that you might be trying to refactor or migrate from.

Comment Out Code and Fix Errors

Another trick that you can do is, when you see the variable or object related to what you know you need to refactor, for example Flask's g, or an import of a library you want to stop using, you can do one trick, just comment out that line in the file.

Then the editor will complain of errors around in the file for things that depended on that, you can scroll to those errors and fix them. It might be the case that you don't even want to remove that variable or that import completely, but temporarily commenting it out to trigger those errors can help guide you directing your attention to the thing you need to change.

The same would work, for example, refactoring request bodies from Flask that would access the request object. You can comment out that line, for example:

# data = request.get_json()

...and that will trigger an error in the next place that used that variable data.

You will probably update that variable data to be a Pydantic model.

And once it is a Pydantic model with type checks, the editor will complain in the places where you try to use it as a dictionary, so you can focus your attention on those places.

Clean Up Tips

And to finish, a couple of tips to keep your code base tidy.

You can use several tools that will help you find errors in your code. I already told you above about Black, Isort, and Autoflake. Those would format the code automatically.

But there are other tools that will help you find errors and many bugs right in the editor.

You can use mypy to check the type annotations. It will help you find errors in your code, for example if you are trying to sum something that could be None and an int, that would throw an error, mypy will help you detect it.

mypy is a command line tool, so, you can run it on continuous integration to make sure you are not adding possible future bugs to your code. And it can also be integrated with all the editors, so that you can get inline errors. And you would get the same errors that you are checking in CI, which is great.

The same way you can use Flake8, that can help you check another range of types of possible errors you could make. You can also try Ruff instead, a tool that includes pretty much the functionality from Flake8 plus some others, but is extremely fast.

There are other linters (the tools that check your code for errors), but some of those could complain about simple things, for example Pylint, like not having a docstring, or the proper formatting for a docstring in some class, etc... and that error ends up mixed with an actual bug, like summing None with an int. So I normally prefer to stick to Flake8 or Ruff for those types of linters as they show the minimum of actual potential errors, to avoid missing actual problems because there's some noise around.

Then you can also try using Pre-Commit. It's a tool that you connect with Git, and every time you try to commit, it can check and automatically fix things like formatting in your code, telling you what needed to change, so that you can add those changes to Git and commit again. It can help you a lot keeping the code consisten through time.

Recap

By this point, you have a very complete overview of how to migrate a large app from Flask to FastAPI.

I would hope you don't even need all this information, but if you have some of these complexities in your code base, then I hope these tips might be useful to you.