FastAPI quickstart (for Flask developers)

Intro

I love Flask, but recently I've switched to using FastAPI for all new side projects.

FastAPI is a high performance modern web framework with first class support for Python 3. The biggest reasons I like it are:

• First class support for Python3 - using type hints for data validation
• Async support
• Uses dependency injection
• Automatic creation of an OpenAPI specification from code
• As simple as Flask

So if you're an existing Flask developer and want to make the switch to FastAPI - this is a small guide highlighting the differences.

Basic routing in FastAPI

You'll notice immediately that in FastAPI the routing decorator includes the method the function handles, whereas in Flask you specify the methods that are handled via the methods=['GET', 'POST'] decorator parameter.

In my opinion this is cleaner. Most projects I've worked do not have a single function that deals with multiple HTTP methods. We typically have individual functions that deal with a single HTTP method for routes. This reduces complexity and makes testing easier.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")  # HTTP method type defined in decorator method
async def hello():
  return { 'message': 'world'}

Path and GET query parameters

Path parameters

Using path parameters in FastAPI is very similar to how you would do it in Flask. One key difference is that in Flask you would also define the type of the path parameter in the route description - for instance @app.route("/project/<int:project_id>")

In FastAPI, you define the type using Python 3 type hints! This is one of the biggest features I like about FastAPI - it has first class support for type hints, and you'll see this being taken advantage of throughout the framework.

Below we define a route that expects a project_id parameter of type int:

from fastapi import FastAPI

app = FastAPI()

@app.get("/project/{project_id}")
async def search_project_users(project_id: int):
  total, users = find_users(project_id, search_term)

  return {
    "users": users,
    "total": total
  }

FastAPI will automatically route any requests where the {project_id} part can be parsed as an integer

Query parameters

With Flask, you'll typically use the request.args method to access query parameters from the request, e.g. request.args.get("searchTerm").

With FastAPI, you have a few options depending on what you want to do. The most straightforward way is to define a new function parameter that isn't in the path. FastAPI will assume that it's part of the query instead, for example:

from fastapi import FastAPI

app = FastAPI()

@app.get("/project/{project_id}")
async def search_project_users(project_id: int, search_term: str):  # search_term is a query parameter
  total, users = find_users(project_id, search_term)

  return {
    "users": users,
    "total": total
  }

If you want to make the query parameter optional, you can define it as search_term: str = None. This provides a type hint that search_term is None by default, for example:

from fastapi import FastAPI

app = FastAPI()

@app.get("/project/{project_id}")
async def search_project_users(project_id: int, search_term: str = None):  # search_term is optional
  total, users = find_users(project_id, search_term)

  return {
    "users": users,
    "total": total
  }

There's also the Query class in FastAPI that allows you to do a bit more with query parameters. For example, in the snippet below we can validate the minimum length of the parameter, and we alias the query parameter "searchTerm" to the function parameter search_term

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/project/{project_id}")
async def search_project_users(
  project_id: int,
  search_term: Query(..., min_length=3, alias="searchTerm")
):
  total, users = find_users(project_id, search_term)

  return {
    "users": users,
    "total": total
  }

Parsing the request body

To get request body data in Flask, you'll typically use request.json() or you can use something like marshmallow to define a schema that maps to the request body. See my article on using flask and marshmallow for how this can be done in Flask.

With FastAPI, we use Pydantic to define the model that the request body maps to. It's similar to how Marshmallow works, but it's faster and uses Python 3 type hints to define model types.

Below is an example of creating a model for a request that contains data to create a new project:

from fastapi import FastAPI, Query
from pydantic import BaseModel, EmailStr, constr

class CreateProject(BaseModel):
  name: constr(min_length=3)
  description: str
  owner_email: EmailStr

app = FastAPI()

@app.post("/project")  # This endpoint only accepts POSTs requests
async def create_project(
  project_data: CreateProject# The request body must match this pydantic model
):
  project = create_new_project(project_data)

  return project

Rather than defining the schema using a decorator, it's defined as the type of the user_data parameter that's passed to the create_user function. FastAPI will automatically workout that that request body should match the schema of the model, and will try to coerce the data into the right format.

If the request data doesn't match the spec of the model, FastAPI will return a 422 validation error

This tight integration between pydantic models and FastAPI leads to much more concise code.

Setting the response model

When building a JSON APIs with Flask, you'll use the jsonify function to serialize your data to send back to the client. You can also use Marshmallow to serialize your objects to dictionaries which can then be sent as a JSON encoded response to the client.

In FastAPI you define a response model using Pydantic. This model defines the "shape" of the data you're sending back to the client. It's identical to how we define what the request body should look like.

When a route has a response model set, FastAPI will serialize the data according to that model and will validate the data to make sure it matches the model.

To specify a response model, you setup the route like this:

from fastapi import FastAPI, Query
from pydantic import BaseModel, EmailStr, constr

# The request model
class CreateProjectSchema(BaseModel):
  name: constr(min_length=3)
  description: str
  owner_email: EmailStr

# The response model, it includes the ID of the newly created project
class ProjectSchema(CreateProjectSchema):
  id: int

app = FastAPI()

# This endpoint must return data that matches the Project model
@app.post("/project", response_model=ProjectSchema)  
async def create_project(
  project_data: CreateProjectSchema  # The request body must match the CreateProject model
):
  project = create_new_project(project_data)

  return project

Switching from Blueprints to APIRouter

In Flask when you need to group endpoints together to create more modular apps you use Blueprints, in FastAPI you have a similar concept called APIRouter.

from fastapi import FastAPI, APIRouter, HTTPException

projects_router = APIRouter(prefix="/projects")

@projects_router.get("/{id}")
async def get_project(id: int):
  project = get_project_by_id(id)
  if not project:
    raise HTTPException(404, "Project not found")
  return project

@projects_router.post("/")
async def create_project(project_data: CreateProject):
  project = create_new_project(project_data)
  return project

app = FastAPI()
app.include_router(projects_router) 

Moving from Blueprints to APIRouter is very straight forward, the interfaces for both are very similar.

Integrating SQLAlchemy

With Flask, you can use the Flask-SQLAlchemy plugin to quickly and easily setup the integration with SQLAlchemy.

With FastAPI however, there is a bit more work involved in getting SQLAlchemy integrated as there's no equivalent plugin. However, because FastAPI makes use of dependency injection, you can integrate with SQLAlchemy in less than a 100 lines of code.

The first step involves setting up the SQLAlchemy engine connection, and the session factory that our application will use:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# import your db connection details, ideally these should be stored in your environment
# variables
from app.config import DATABASE_HOST, DATABASE_NAME, DATABASE_PASSWORD, DATABASE_USER

# helper function to create a DSN with our connection details
def get_url():
    return "postgresql://%s:%s@%s/%s" % (
        DATABASE_USER,
        DATABASE_PASSWORD,
        DATABASE_HOST,
        DATABASE_NAME,
    )

# this creates our connection to postgres
engine = create_engine(get_url())

# Session will be used to construct the sqlalchemy Session when we're interacting
# with the database 
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# This will be inherited by our sqlalchemy ORM models
Base = declarative_base()

# This will be used as a dependenacy in our FastAPI routes when we need a session
# to interact with the database
async def get_db():
    try:
        db = Session()
        yield db
    finally:
        db.close()

The next step is to use the above get_db function as a dependency in our FastAPI routes whenever we need a connection to the database:

from sqlalchemy.orm import Session
from fastapi import FastAPI, APIRouter, HTTPException
from .db import get_db  # import from the file above
from .models import get_project_by_id

app = FastAPI()

@app.get("/projects/{id}")
async def get_project(id: int, db: Session = Depends(get_db)):
  project = get_project_by_id(db, id)
  if not project:
    raise HTTPException(404, "Project not found")
  return project

The key here is that we add a new parameter into our route called db which FastAPI will inject our SQLAlchemy session into using Depends(get_db).

Dependency injection a technique that's used by FastAPI throughtout the framework. You've seen it in action when defining path and query parameters, and now we've used it for our database connection.

We also give the db parameter a Session typehint so that we get code completion in editors that support it.

Another important difference is that in FastAPI whenever a function needs access to the current request's SQLAlchemy session, we have to pass it in via a function parameter. There is no notion of a global Session instance that's tied to the current request context like we have with the Flask-SQLAlchemy plugin.

To flesh this point out, here's how we could construct our get_project_by_id function using the Session object that was injected by FastAPI

from datetime import datetime
from sqlalchemy.orm import Session
from sqlalchemy import (
    Column,
    DateTime,
    Integer,
    UnicodeText,
)
from .db import Base

class Project(Base):
    __tablename__ = "project"

    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(UnicodeText, nullable=False)
    created_at = Column(DateTime, nullable=False, default=datetime.now)


def get_project_by_id(session: Session, id: int) -> Project:
  return session.query(Project).filter(Project.id == id).first()

Basic user authentication

With Flask it's relatively straightforward to get user session management integrated into your application using popular plugins like Flask-Security, Flask-Login, etc.

However in FastAPI you can roll your own authentication system, or use the fastapi.security module to implement OAuth2 based security.

In this example, we're going to implement our own security system that will check requests for an Authorization header to confirm that that requests are authenticated. This is what's typically used for pure API based backends.

We're first going to create a function which will be used as a dependency in our routing functions.

from fastapi import Header, Depends, HTTPException
from .db import get_db
from starlette.status import HTTP_401_UNAUTHORIZED
from .models import get_user_by_auth_token


async def get_current_user(
  db: None = Depends(get_db),  # Our database dependency
  authorization: str = Header(None)  # Injected by FastAPI when this function is used in a route
):
    auth_token = None
    if authorization and "bearer" in authorization.lower():
        scheme, _, auth_token = authorization.partition(" ")

    if not auth_token:
        raise HTTPException(HTTP_401_UNAUTHORIZED, "You must be logged in")

    # retrieve the user from the database
    user = get_user_by_auth_token(db, auth_token)
    if not user:
        raise HTTPException(HTTP_401_UNAUTHORIZED, "Invalid token")

    return user

The above function can use used as a dependency in a FastAPI route function:

from sqlalchemy.orm import Session
from fastapi import FastAPI, APIRouter, HTTPException
from .db import get_db
from .models import get_project_by_id, User
from .auth import get_current_user

app = FastAPI()

@app.get("/projects/{id}")
async def get_project(
  id: int,
  db: Session = Depends(get_db),
  user: User = Depends(get_current_user)  # This route now requires an authenticated user
):
  project = get_project_by_id(db, id)
  if not project:
    raise HTTPException(404, "Project not found")
  return { "project": project }

FastAPI will automatically inject the db and the authorization parameters for the get_current_user function. This enables us to compose functions that have multiple sub-dependencies.

Further reading

That was a short tour of how FastAPI compares to Flask. To get more information about FastAPI, I recommend these links:

• The FastAPI official documentation
• pydantic documentation