High-performing Apps With Python: A FastAPI Tutorial

Good programming language frameworks make it easy to produce quality products faster. Great frameworks even make the whole development experience enjoyable. FastAPI is a new Python web framework that’s powerful and enjoyable to use. The following features make the Python FastAPI framework worth trying:

• Speed: FastAPI is one of the fastest Python web frameworks. In fact, its speed is at par with Node.js and Go. Check these FastAPI performance tests.
• The FastAPI documentation is detailed and easy-to-use.
• Type hint your code and get free data validation and conversion.
• Create plugins easily using dependency injection.

Python FastAPI Tutorial: Building a TODO App

To explore the big ideas behind FastAPI, let’s build a TODO app, which sets up to-do lists for its users. Our FastAPI example app will provide the following features:

• Signup and Login
• Add new TODO item
• Get a list of all TODOs
• Delete/Update a TODO item

SQLAlchemy for Data Models

Our app has just two models: User and TODO. With the help of SQLAlchemy, the database toolkit for Python, we can express our models like this:

class User(Base):
   __tablename__ = "users"
   id = Column(Integer, primary_key=True, index=True)
   lname = Column(String)
   fname = Column(String)
   email = Column(String, unique=True, index=True)
   todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan")
 
class TODO(Base):
   __tablename__ = "todos"
   id = Column(Integer, primary_key=True, index=True)
   text = Column(String, index=True)
   completed = Column(Boolean, default=False)
   owner_id = Column(Integer, ForeignKey("users.id"))
   owner = relationship("User", back_populates="todos")

Once our models are ready, let’s write the configuration file for SQLAlchemy so that it knows how to establish a connection with the database.

import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL']
engine = create_engine(
   SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

Unleash the Power of Type Hints

A sizable part of any API project concerns the routine stuff like data validation and conversion. Let’s tackle it up front before we go to writing request handlers. With FastAPI, we express the schema of our incoming/outgoing data using pydantic models and then use these pydantic models to type hint and enjoy free data validation and conversion. Please note these models are not related to our database workflow and only specify the shape of data that’s flowing in and out of our REST interface. To write pydantic models, think about all the ways the User and TODO information will flow in and out.

Traditionally, a new user will sign up for our TODO service and an existing user will log in. Both of these interactions deal with User information, but the shape of data will be different. We need more information from users during signup and minimal (only email and password) when logging in. This means we need two pydantic models to express these two different shapes of User info.

In our TODO app, however, we will leverage the built-in OAuth2 support in FastAPI for a JSON Web Tokens (JWT)-based login flow. We just need to define a UserCreate schema here to specify data that will flow into our signup endpoint and a UserBase schema to return as a response in case the signup process is successful.

from pydantic import BaseModel
from pydantic import EmailStr
class UserBase(BaseModel):
   email: EmailStr
class UserCreate(UserBase):
   lname: str
   fname: str
   password: str

Here, we marked last name, first name, and password as a string, but it can be further tightened by using pydantic constrained strings that enable checks such as min length, max length, and regexes.

To support the creation and listing of TODO items, we define the following schema:

class TODOCreate(BaseModel):
   text: str
   completed: bool

To support the update of an existing TODO item, we define another schema:

class TODOUpdate(TODOCreate):
   id: int

With this, we are done with defining schemas for all data exchanges. We now turn our attention to request handlers where these schemas will be used to do all the heavy lifting of data conversion and validation for free.

Let Users Sign Up

First, let’s allow users to sign up, as all of our services need to be accessed by an authenticated user. We write our first request handler using the UserCreate and UserBase schema defined above.

@app.post("/api/users", response_model=schemas.User)
def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)):
   """add new user"""
   user = crud.get_user_by_email(db, user_data.email)
   if user:
   	raise HTTPException(status_code=409,
   	                    detail="Email already registered.")
   signedup_user = crud.create_user(db, user_data)
   return signedup_user

There’s a lot going on in this short piece of code. We have used a decorator to specify the HTTP verb, the URI, and the schema of successful responses. In order to ensure that the user has submitted the right data, we have typed hint the request body with an earlier defined UserCreate schema. The method defines another parameter for getting a handle on the database—this is dependency injection in action and is discussed later in this FastAPI tutorial.

Securing Our API

We want the following security features in our FastAPI example:

• Password hashing
• JWT-based authentication

For password hashing, we can use Passlib. Let’s define functions that handle password hashing and checking if a password is correct.

from passlib.context import CryptContext
 
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
def verify_password(plain_password, hashed_password):
   return pwd_context.verify(plain_password, hashed_password)
 
 
def get_password_hash(password):
   return pwd_context.hash(password)
 
def authenticate_user(db, email: str, password: str):
   user = crud.get_user_by_email(db, email)
   if not user:
   	return False
   if not verify_password(password, user.hashed_password):
   	return False
   return user

To enable JWT-based authentication, we need to generate JWTs as well as decode them to get user credentials. We define the following functions to provide this functionality.

# install PyJWT
import jwt
from fastapi.security import OAuth2PasswordBearer
 
SECRET_KEY = os.environ['SECRET_KEY']
ALGORITHM = os.environ['ALGORITHM']
 
def create_access_token(*, data: dict, expires_delta: timedelta = None):
   to_encode = data.copy()
   if expires_delta:
   	expire = datetime.utcnow() + expires_delta
   else:
   	expire = datetime.utcnow() + timedelta(minutes=15)
   to_encode.update({"exp": expire})
   encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
   return encoded_jwt
def decode_access_token(db, token):
   credentials_exception = HTTPException(
   	status_code=HTTP_401_UNAUTHORIZED,
   	detail="Could not validate credentials",
   	headers={"WWW-Authenticate": "Bearer"},
   )
   try:
   	payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
   	email: str = payload.get("sub")
   	if email is None:
       	raise credentials_exception
   	token_data = schemas.TokenData(email=email)
   except PyJWTError:
   	raise credentials_exception
   user = crud.get_user_by_email(db, email=token_data.email)
   if user is None:
   	raise credentials_exception
   return user

Issue Tokens at Successful Login

Now, we will define a Login endpoint and implement the OAuth2 password flow. This endpoint will receive an email and password. We will check the credentials against the database, and on success, issue a JSON web token to the user.

To receive the credentials, we will make use of OAuth2PasswordRequestForm, which is a part of FastAPI’s security utilities.

@app.post("/api/token", response_model=schemas.Token)
def login_for_access_token(db: Session = Depends(get_db),
                      	form_data: OAuth2PasswordRequestForm = Depends()):
   """generate access token for valid credentials"""
   user = authenticate_user(db, form_data.username, form_data.password)
   if not user:
   	raise HTTPException(
       	status_code=HTTP_401_UNAUTHORIZED,
       	detail="Incorrect email or password",
       	headers={"WWW-Authenticate": "Bearer"},
   	)
   access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
   access_token = create_access_token(data={"sub": user.email},
                                  	expires_delta=access_token_expires)
   return {"access_token": access_token, "token_type": "bearer"}

Using Dependency Injection to Access DB and Protect Endpoints

We have set up the login endpoint that provides a JWT to a user upon successful login. The user can save this token in local storage and show it to our back end as an Authorization header. The endpoints that expect access only from logged-in users can decode the token and find out who the requester is. This kind of work is not tied to a particular endpoint, rather it’s shared logic utilized in all protected endpoints. It’s best to set up the token-decoding logic as a dependency that can be used in any request handler.

In FastAPI-speak, our path operation functions (request handlers) would then depend on get_current_user. The get_current_user dependency needs to have a connection to the database and to hook into the FastAPI’s OAuth2PasswordBearer logic to obtain a token. We will resolve this issue by making get_current_user depend on other functions. This way, we can define dependency chains, which is a very powerful concept.

def get_db():
   """provide db session to path operation functions"""
   try:
   	db = SessionLocal()
   	yield db
   finally:
   	db.close()
def get_current_user(db: Session = Depends(get_db),
                	token: str = Depends(oauth2_scheme)):
   return decode_access_token(db, token)
@app.get("/api/me", response_model=schemas.User)
def read_logged_in_user(current_user: models.User = Depends(get_current_user)):
   """return user settings for current user"""
   return current_user

Logged-in Users Can CRUD TODOs

Before we write the path operation functions for TODO Create, Read, Update, Delete (CRUD), we define the following helper functions to perform actual CRUD on the db.

def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate):
   todo = models.TODO(text=todo_data.text,
                   	completed=todo_data.completed)
   todo.owner = current_user
   db.add(todo)
   db.commit()
   db.refresh(todo)
   return todo
def update_todo(db: Session, todo_data: schemas.TODOUpdate):
   todo = db.query(models.TODO).filter(models.TODO.id == id).first()
   todo.text = todo_data.text
   todo.completed = todo.completed
   db.commit()
   db.refresh(todo)
   return todo
def delete_todo(db: Session, id: int):
   todo = db.query(models.TODO).filter(models.TODO.id == id).first()
   db.delete(todo)
   db.commit()
 
def get_user_todos(db: Session, userid: int):
   return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

These db-level functions will be used in the following REST endpoints:

@app.get("/api/mytodos", response_model=List[schemas.TODO])
def get_own_todos(current_user: models.User = Depends(get_current_user),
             	db: Session = Depends(get_db)):
   """return a list of TODOs owned by current user"""
   todos = crud.get_user_todos(db, current_user.id)
   return todos
@app.post("/api/todos", response_model=schemas.TODO)
def add_a_todo(todo_data: schemas.TODOCreate,
          	current_user: models.User = Depends(get_current_user),
          	db: Session = Depends(get_db)):
   """add a TODO"""
   todo = crud.create_meal(db, current_user, meal_data)
   return todo
@app.put("/api/todos/{todo_id}", response_model=schemas.TODO)
def update_a_todo(todo_id: int,
             	todo_data: schemas.TODOUpdate,
             	current_user: models.User = Depends(get_current_user),
             	db: Session = Depends(get_db)):
   """update and return TODO for given id"""
   todo = crud.get_todo(db, todo_id)
   updated_todo = crud.update_todo(db, todo_id, todo_data)
   return updated_todo
@app.delete("/api/todos/{todo_id}")
def delete_a_meal(todo_id: int,
             	current_user: models.User = Depends(get_current_user),
             	db: Session = Depends(get_db)):
   """delete TODO of given id"""
   crud.delete_meal(db, todo_id)
   return {"detail": "TODO Deleted"}

Write Tests

Let’s write a few tests for our TODO API. FastAPI provides a TestClient class that’s based on the popular Requests library, and we can run the tests with Pytest.

To make sure only logged-in users can create a TODO, we can write something like this:

from starlette.testclient import TestClient
from .main import app
client = TestClient(app)
def test_unauthenticated_user_cant_create_todos():   todo=dict(text="run a mile", completed=False)
response = client.post("/api/todos", data=todo)
assert response.status_code == 401

The following test checks our login endpoint and generates a JWT if presented with valid login credentials.

def test_user_can_obtain_auth_token():
  response = client.post("/api/token", data=good_credentials)
  assert response.status_code == 200
  assert 'access_token' in response.json()
  assert 'token_type' in response.json()

Summing It Up

We’re done implementing a very simple TODO app using FastAPI. By now, you’ve seen the power of type hints put to good use in defining the shape of incoming and outgoing data through our REST interface. We define the schemas at one place and leave it to FastAPI to apply data validation and conversion. The other noteworthy feature is dependency injection. We used this concept to package the shared logic of obtaining a database connection, decoding the JWT to get the currently logged-in user, and implementing simple OAuth2 with password and bearer. We also saw how dependencies can be chained together.

We can easily apply this concept to add features such as role-based access. FastAPI performance and documentation are second to none, making it easy to master and use. We are writing concise and powerful code without learning the peculiarities of a framework. In simple words, FastAPI is a collection of powerful tools that you don’t have to learn because they are just modern Python. Have fun.