Intro

At Lambert Labs we are always interested in trying out new Python frameworks that are well built and managed. We recently started using a framework called FastAPI. FastAPI is a server-side Python framework used to build APIs for consumption by various different clients. As the name suggests, FastAPI is high-performance – it is regarded as one of the fastest Python frameworks available. In fact, according to tests run by www.techempower.com, FastAPI framework outperforms other popular Python web frameworks including Django and Flask!

In this particular test, all the frameworks were put through their paces fetching data from a database that they have no prior knowledge of. They had to read, modify and sort the data they fetched as quickly as possible and the faster they do this, the higher performance of the framework.

What is FastAPI?

As stated above, FastAPI is a framework used to make API services, which will be consumed by users. It is written to be coded in Python 3.6+. According to FastAPI’s creator, the framework was designed to implement features that take advantage of Python 3.6+ based features (type hints, for example) and be detailed and easy to use to make the developer experience smooth. This is not to mention the performance, which as one can see from the test mentioned above, is excellent.

Explaining FastAPI

Many of the frameworks that _appear_ faster than FastAPI aren’t directly comparable because of the difference in the features they offer. Frameworks like Sanic and Starlette (FastAPI is based on Starlette) do not have the full data validation or JSON serialization features that FastAPI offers and coding them manually would introduce the same (or more) overhead to FastAPI. Therefore, when you narrow the comparison of FastAPI to other fleshed out web frameworks like Django and Flask, you can see why FastAPI is increasing in popularity.

Data validation

As alluded to above, one of the many great features of FastAPI is data validation. This is due to Pydantic; the second dependency that FastAPI is built on top of (in addition to Starlette). Pydantic enforces data types during the application’s run time. It ensures that when a consumer of an API endpoint sends data to the server but the consumer has incorrectly sent the wrong type of data, then the server can respond with helpful error messages, instead of attempting to map the data to the database and potentially causing an I/O operation failure.

Built on standards

FastAPI is robust because of the standards it adheres to and uses; namely OpenAPI and JSONSchema. OpenAPI is a widely adopted specification for defining a language-agnostic standard for creating APIs. This includes standards defining path operations, security dependencies, query parameters and more. Adopting a common standard that is widely known by other developers allows FastAPI applications to be more scalable and makes the development experience a more consistent one. 

JSONSchema validates JSON data and describes the appropriate format to use in endpoint requests and responses. Combined with OpenAPI, FastAPI leverages these standards to create automatic API documentation so that developers can consume the APIs in a web interface: Swagger UI or Redoc. Having Swagger UI or Redoc available in a developer’s toolbox is essential for performing quick sanity checks on a particular endpoint – it helps to replicate the frontend application experience.

Walkthrough: Authorization scopes

The best way to demonstrate FastAPI is walking through an implementation of commonly used features. I have chosen to take an advanced feature from the FastAPI documentation and one which we at Lambert Labs have recently adapted and are using in current projects. In particular, I am going to demonstrate how to add authorization ‘scopes’ to an endpoint in FastAPI.

Authorization scopes are specific, granular permissions which are given to users of an application to ensure that they aren’t given privileged access to certain features of the application that they shouldn’t have access to. They can also be part of role policies which a human or machine can assume temporarily depending on their use case. Either way, it helps to distinguish levels of access to ensure that the right people have the appropriate permissions at all times. Take Instagram as an example. There are a variety of different permissions that can be given to consumers of the API. For instance, there are scopes for getting or editing a user’s profile picture, but these scopes are kept separate so that users can only view other users’ profile pictures but can only edit their own.

Walkthrough

Let’s walk through the example implementation mentioned in the FastAPI documentation.

from datetime import datetime, timedelta
from typing import List, Optional
from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (OAuth2PasswordBearer,
                              OAuth2PasswordRequestForm,
		              SecurityScopes)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

fake_users_db =
{"johndoe":
    {"username": "johndoe"
     "full_name": "John Doe"
     "email": "johndoe@example.com",
     "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
    },
 "alice":
    {"username": "alice",
     "full_name": "Alice Chains",
     "email": "alicechains@example.com",
     "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
    },
}

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None
    scopes: List[str] = []

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None

class UserInDB(User):
    hashed_password: str

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token",
				     scopes={"me": "Read information about the current user.",
					     "items": "Read items."},)

app = FastAPI()

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 get_user(db, username: str):
    if username in db:
	user_dict = db[username]
	return UserInDB(**user_dict)

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db,  username)
    if not user:
	return False
    if not verify_password(password,  user.hashed_password):
	return False
    return user

def create_access_token(data: dict, expires_delta: Optional[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

async def get_current_user(security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)):
    if security_scopes.scopes:
	authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
	authenticate_value = f"Bearer"

    credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
					  detail="Could not validate credentials",
					  headers={"WWW-Authenticate": authenticate_value},)
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
	username: str = payload.get("sub")
	if username is None:
		raise credentials_exception
	token_scopes = payload.get("scopes", [])
	token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
	raise credentials_exception

    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
	if scope not in token_data.scopes:
	    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
			        detail="Not enough permissions",
				headers={"WWW-Authenticate": authenticate_value},)
    return user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
	raise HTTPException(status_code=400,
		            detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(data={"sub": user.username, "scopes": form_data.scopes},
				       expires_delta=access_token_expires,)

    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/items/")
async def read_own_items(current_user: User = Security(get_current_user, scopes=["items"])):
    return [{"item_id": "Foo", "owner": current_user.username}]

The first few lines of the code import dependencies used in the application. There is also a small example database called fake_users_db, which is just a dictionary implementation of a database used  for demonstration purposes. The class schemas Token, TokenData, User and UserInDB are Python classes that validate data when it is sent to the API via HTTP or when it is about to be returned to the consumer of the API via HTTP . Notice that `UserInDB` has the same attributes as the ‘column’ data in fake_users_db (i.e. username, hashed_password, etc). This class would be used to validate any request data that describes a user to check that it has the same attributes/data type as those in the user database. pwd_context is a helper for hashing and unhashing passwords used in the database. oauth2_scheme is a very simple security dependency. All oauth2_scheme does is that it checks that the Authorization header in a request contains a JWT token (explained more below). Everything else defined below is what might be considered the ‘actual’ application workflow.

Purpose

Whilst the application code contains a lot of security ‘logic’, including but not limited to authorization scopes. It would not make sense to present authorization scopes in isolation, without any of the other security that should go along with it. Hence, whilst the focus of this blog is on Authorization scopes, one can see a typical FastAPI implementation of other security features as well. Refer to the flowchart below for a simplified view of how authentication fits into an API application.

As stated above, the purpose of the code is a full authentication workflow: checking a database against the credentials given to a user, assigning a temporary access token they use to consume endpoints, decoding and validating that token when a consumer of an endpoint submits it as part of a request. On top of that, it also has a basic implementation of the authorization scopes. All things considered, it is rather simple.

Granting access token

Let’s examine a user requesting a temporary access token and how that is handled:

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
	raise HTTPException(status_code=400,
		            detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(data={"sub": user.username, "scopes": form_data.scopes},
				       expires_delta=access_token_expires,)

    return {"access_token": access_token, "token_type": "bearer"}

From the first line, we can see that any request made to this endpoint must have a body which conforms to a special type – OAuth2PasswordRequestForm. This class is provided by FastAPI – to save you time writing your own. OAuth2PasswordRequestForm has commonly-used attributes such as ‘username’, ‘password’ and ‘scope’. After checking in the database that the user exists, an access token is created for the user. The access token consists of data describing the user, their access time limits and the scope permissions assigned to them that is encoded into a compact string-type object, which is the token. A popular encoded access token is JWT, which is a standard for encrypting JSON information used in authentication/authorization. In this example, the user themselves define which permission scope they wanted when they made the request for the access token. In production, this would be done in the database where the user’s role in the application would be recognised (i.e. user, admin, developer, etc).

Scopes in action

The interesting stuff happens when the user attempts to consume endpoints with the access token they just received.

@app.get("/users/me/items/")
async def read_own_items(current_user: User = Security(get_current_user, scopes=["items"])):
    return [{"item_id": "Foo", "owner": current_user.username}]

In a fully authenticated example, the end response from this endpoint is the user’s username and a key item_id with value Foo. In the path operation, one can see that there is a Security dependency on the path operation called get_current_user. This dependency is measured against the endpoint specific scope, items. The idea is that only users who are granted the items scope can consume this endpoint and get the desired response.

async def get_current_user(security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)):
    if security_scopes.scopes:
	authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
	authenticate_value = f"Bearer"

    credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
					  detail="Could not validate credentials",
					  headers={"WWW-Authenticate": authenticate_value},)
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
	username: str = payload.get("sub")
	if username is None:
		raise credentials_exception
	token_scopes = payload.get("scopes", [])
	token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
	raise credentials_exception

    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
	if scope not in token_data.scopes:
	    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
			        detail="Not enough permissions",
				headers={"WWW-Authenticate": authenticate_value},)
    return user

Let’s examine the dependencies. The first dependency is SecurityScopes, which contains the endpoint’s required permission scopes. As a reminder, there’s only one scope for this endpoint: items. Essentially, this function decodes the user’s JWT back into an object containing the user’s name and the scopes that they were granted when they first received the access token. After decoding the JWT, the function does three checks on the decoded data: (1) check the user has a username; (2) check that this username exists in the database and (3) check that the required scopes are at least a subset of the scopes granted to the user (in our case, does the user have the permission scope ‘items’?). If (1) or (2), the endpoint returns an error response with status code 401 (Unauthorized) and if (3) should fail, the endpoint returns an error response with status code 403 (Forbidden).

If the checks do not throw any errors, the endpoint response data is returned and the request is a success. It is important to note that the scope items was unique to this endpoint and it is totally possible to define unique scopes or combinations of scopes for different endpoints. The only things that would require changing are the dependencies attached to the path operation function and for the user to change the scopes that they ask for when they sign in. As stated previously, in reality one would first verify a user’s role in a database and assign the scopes that they are permitted to have.

Want to know more about how Lambert Labs use FastAPI in production-grade applications? Check out our FastAPI page.