lou

Posted on Feb 14 • Edited on Feb 16

A Step by Step Guide to Building Lightning Fast APIs

#fastapi #python #pydantic #jwt

FastAPI is a Python framework that lets you build APIs with speed and simplicity. It supports async programming and that's what makes it fast.
In this guide, I’ll walk you through creating your first FastAPI project, enhancing it with data models, integrating a database, and even securing your endpoints with JWT authentication.

A Dummy FastAPI Project

Let’s kick things off with a basic API:

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World"}


@app.get("/hello/{name}")
async def say_hello(name: str):
    return {"message": f"Hello {name}"}

You might be wondering, what’s happening here?
We Created:

An instance of the class FastApi from the module fastapi.
A variable, app, which will serve as the point of interaction to create the api.
Used decorator, @app followed by the HTTP method, get, put, post, delete
Passed the endpoint to the decorator and defined the operation function

Try it out by running your server and visiting

http://127.0.0.1:8000/docs

To explore the interactive documentation

Your First FastAPI Project

Now let's create a model to hold the schema of our data type. There's one problem with using plain Python classes for this: FastAPI won't know how to handle these models automatically, meaning you'll have to manually parse and validate the request body. To avoid this, use Pydantic's BaseModel as recommended in their documentation.

Let's explore how to add a Starship to our API by defining its data type using a Pydantic model:

First, we import BaseModel from Pydantic.
Then, we define the Starship class by inheriting from BaseModel and listing out the attributes.
This method automates data validation, making the API easy to maintain.

	from fastapi import FastAPI
	from pydantic import BaseModel

	app = FastAPI()

	class Starship(BaseModel):
	id: str
	name: str
	price: float
	discount: float \| None

	@app.post("/addStarship")
	async def add_starship(starship: Starship):
	return {
	"starship_id": starship.id,
	"starship": starship
	}

view raw add_starship_fastapi.py hosted with ❤ by GitHub

We’ll dive into the endpoint creation details as we move along. For now launch the following URL to test things out:

http://127.0.0.1:8000/docs#

This interactive docs page lets you see your endpoints in action. All you have to do is simply add the required details and watch the results:

Now, let’s spice things up by adding a Jedi to link each Starship to its purchaser. First, we need to define the fields. Pydantic’s Field is a lifesaver here, it not only lets you add extra information to a field but can also use a default_factory to auto-generate values. This is perfect for automatically creating an ID whenever a new Starship or Jedi is instantiated.

Additionally, you can use Field to make your API documentation richer by adding descriptions. For more details on the available parameters, check out the Pydantic documentation

	class Jedi(BaseModel):
	id: str = Field(default_factory=lambda: f"schorle-{uuid.uuid4()}", exclude=True)
	name: str
	home_world: str = Field(...,
	title="Homeworld",
	description="Homeworld Also spelled home world or known as home planet. An individual's homeworld could be their place of birth. Dathomir was the homeworld of Maul and Asajj Ventress",
	example="Dathomir")

view raw jedi_class_fastapi.py hosted with ❤ by GitHub

Note the use of ellipsis within Field. It emphasize on the fact that a value must be provided.

Schema Definition

Since the ID is auto generated, having a well defined schema helps prevent unnecessary inputs from the client. In Pydantic v2, you should now define the JSON schema using model_config = ConfigDict() instead of the deprecated json_schema_extra.

For fields that are computed from other fields, simply add the @computed_field decorator to the property function. It’s recommended to explicitly use the @property decorator, but it isn’t required.

	from fastapi import FastAPI
	from pydantic import BaseModel, computed_field, Field, ConfigDict
	import uuid

	app = FastAPI()

	class Starship(BaseModel):
	id: str = Field(default_factory=lambda: f"schorle-{uuid.uuid4()}", exclude=True)
	name: str
	price: float
	discount: float \| None
	model_config = ConfigDict(json_schema_extra={
	"example": {
	"name": "X-wing",
	"price": 100000,
	"discount": 10
	}
	})
	@computed_field
	@property
	def total(self) -> float:
	return self.price - (self.price * self.discount / 100)
	class Jedi(BaseModel):
	id: str = Field(default_factory=lambda: f"schorle-{uuid.uuid4()}", exclude=True)
	name: str
	home_world: str
	model_config = ConfigDict(json_schema_extra={
	"example": {
	"name": "Luke Skywalker",
	"home_world": "Tatooine"
	}
	})

	@app.post("/addStarship")
	async def add_starship(starship: Starship):
	return {
	"starship_id": starship.id,
	"starship": starship
	}

	@app.post("/purchase")
	async def purchase(starship: Starship, jedi: Jedi):
	return {
	"starship": starship,
	"jedi": jedi
	}

view raw purchase.fastapi.py hosted with ❤ by GitHub

Now let's explore some more Pydantic types, and add them to our code.

HttpUrl is a type that accepts http or https URLs.

We will be using it to provide a list of Photos to show all of the angles of the Starship for a 360 view.

	from pydantic import BaseModel, HttpUrl

	class Photo(BaseModel):
	url: HttpUrl

view raw http_url_pydantic.py hosted with ❤ by GitHub

We will be using Form for sign in purposes

	from fastapi import FastAPI, Form
	app = FastAPI()

	@app.post("/signin")
	def login(email: str = Form(...), password: str = Form(...)):
	return {"email": email}

view raw signin_fastapi.py hosted with ❤ by GitHub

Let's try it out

Awesome!

Integrating a Database with SQLAlchemy and SQLite

Now that we want to save the data sent by the client, it's time to integrate a database into our app. We'll walk through the steps of using SQLAlchemy—a popular ORM for Python—together with SQLite. We’re keeping it simple: create a new directory (let’s call it starship_db), set up a virtual environment if you haven’t already, activate it, and run the following command to get started.

pip3 install sqlalchemy fastapi pydantic

Let's set up your project files with a few quick terminal commands. This step creates your SQLite database file and organizes your project structure:

touch starship.db
mkdir starship
cd starship
touch main.py database.py models.py schema.py

The models.py will hold your models, each model is a blueprint for a database table.

	from sqlalchemy import Column, Float, String
	from .database import Base
	class Starship(Base):
	__tablename__ = 'starship'
	id = Column(String, primary_key=True)
	name = Column(String)
	price = Column(Float)
	discount = Column(Float)

view raw models_fastapi.py hosted with ❤ by GitHub

Here the Starship class serves as a blueprint for a database table. By setting tablename = 'starship' you're instructing SQLAlchemy to create a table in your database named "starship"

Next take your existing Starship model and add it into your schema.py file to define a response model called StarshipResponse. In this model, make sure to include orm_mode=True this tells FastAPI to serialize your SQLAlchemy objects to JSON, ensuring smooth and accurate API responses.


	from pydantic import BaseModel
	class Starship(BaseModel):
	id: str
	name: str
	price: float
	discount: float \| None

	class StarshipResponse(BaseModel):
	id: str
	name: str
	price: float
	discount: float \| None
	class Config:
	orm_mode = True

view raw schema_fastapi.py hosted with ❤ by GitHub

Managing Database Migrations with Alembic

We’ll use Alembic as our go to migration tool to manage changes to our database schema and keep everything running smoothly as our code grows. Alembic makes it much easier to handle schema updates without the headache of manual migrations.

You need to install Alembic and initialize it

pip install alembic
alembic init alembic

This command will create a new directory named alembic in your project and an alembic.ini file. Inside the alembic directory, there will be env.py file and a versions directory for your migration scripts.

Inside of your alembic.ini file, update sqlalchemy.url
then go to alembic/env.py and change this line

target_metadata = None

to your model's metadata, something like this:

from starship.database import Base
target_metadata = Base.metadata

This will link Alembic with your models and allows it to detect any changes in the models when generating migration scripts.

Initialize the migration with:

alembic revision --autogenerate -m "Init"

And apply it with:

alembic upgrade head

Now let's set up our database connection and session, and define a helper function to yield a session instance. This ensures that your database connections are managed properly, cleaning up after each request:

	from sqlalchemy import create_engine, engine
	from sqlalchemy.ext.declarative import declarative_base
	from sqlalchemy.orm import sessionmaker
	Base = declarative_base()
	SQLALCHEMY_DATABASE_URI = 'sqlite:///./starship.db'

	engine= create_engine(SQLALCHEMY_DATABASE_URI,
	connect_args={"check_same_thread": False})
	session = sessionmaker(bind=engine, autocommit=False, autoflush=False)

	def get_db():
	db = session()
	try:
	yield db
	finally:
	db.close()

view raw database_fastapi.py hosted with ❤ by GitHub

Defining CRUD Operations for Starships

In your main.py, you'll use the get_db() method to obtain a database session, and then define a POST endpoint to create a new Starship. This endpoint takes the input object, instantiates the model, adds the new entry to the database, commits the transaction, and finally returns the created record. This approach is fundamental for populating your database via API requests.

db.add(starship)
db.commit()
db.refresh(starship)

And just as we define a POST method to add data. We need to define a PUT, GET and DELETE methods for the rest of the operations. After all, managing your data means you'll also need to update, retrieve, and remove entries as needed.
For these operations we will use filter() to search for a specific starship in the database.


	from fastapi import Depends, APIRouter, HTTPException
	from sqlalchemy.orm import Session
	from . import models
	from starship.database import engine, get_db
	from starship.schemas import Starship

	app = FastAPI()

	models.Base.metadata.create_all(engine)

	@app.post('/starship')
	def add_starship(request: Starship, db: Session = Depends(get_db)):
	starship = models.Starship(
	id= request.id,
	name=request.name,
	price=request.price,
	discount=request.discount
	)
	db.add(starship)
	db.commit()
	db.refresh(starship)
	return starship


	@app.get('/starships')
	def get_starships(db: Session = Depends(get_db)):
	return db.query(models.Starship).all()

	@app.get('/starships/{id}')
	def get_starship(id: str, db: Session =
	Depends(get_db)):
	return db.query(models.Starship).filter(models.Starship.id == id).first()

	@app.delete('/starship/{id}')
	def delete_starship(id: str, db: Session = Depends(get_db)):
	db.query(models.Starship).filter(models.Starship.id == id).delete()
	db.commit()
	return "Starship deleted successfully"

	@app.put('/starship/{id}')
	def update_starship(id: str, request: Starship, db: Session =Depends(get_db)):
	starship = db.query(models.Starship).filter(models.Starship.id == id)
	if not starship:
	raise HTTPException(status_code=404)

	starship.update(request.dict())
	db.commit()
	return "Starship updated successfully"

view raw main_fastapi_db.py hosted with ❤ by GitHub

Now if we try it out:

Defining Relationships: Connecting Starships to Their Pilots

Every starship needs a pilot. Let's create a User model to represent the pilots who navigate these ships and implement the same CRUD operations as we did for Starships.

	class User(BaseModel):
	id: str
	email: str
	password: str

	class UserResponse(BaseModel):
	email: str
	class Config:
	orm_mode = True

view raw user.schemas.fastapi.py hosted with ❤ by GitHub

	class User(Base):
	__tablename__ = 'user'
	id = Column(String, primary_key=True, index= True)
	email = Column(String)
	password = Column(String)

view raw user.model.fastapi.py hosted with ❤ by GitHub

	from typing import List
	from passlib.context import CryptContext
	from fastapi import Depends, APIRouter, HTTPException
	from sqlalchemy.orm import Session
	from . import models
	from starship.database import engine, get_db
	from starship.schemas import Starship, StarshipResponse, UserResponse, User

	app = FastAPI(
	title="USERS API", description="All CRUD operations for Users"
	)

	models.Base.metadata.create_all(engine)
	pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

	@app.post('/user',tags=['users'])
	def create_user(
	request: User, db: Session = Depends(get_db)):
	user= models.User(
	id=request.id,
	email=request.email,
	password= pwd_context.hash(request.password)
	)
	db.add(user)
	db.commit()
	db.refresh(user)
	return user

	@app.get('/user/{id}',tags=['users'])
	def get_user(id: str, db: Session =
	Depends(get_db)):
	return db.query(models.User).filter(models.User.id == id).first()

	@app.delete('/user/{id}', tags=['users'])
	def delete_user(
	id: int, db: Session = Depends(get_db)):
	db.query(models.User).filter(models.User.id == id).delete()
	db.commit()
	return "User deleted successfully"

	@app.get('/users', response_model=List[UserResponse], tags=['users'])
	def get_users(db: Session = Depends(get_db)):
	return db.query(models.User).all()
	@app.put('/user/{id}', tags=['users'])
	def update_user(id: str, request: User, db: Session = Depends(get_db)):
	user = db.query(models.User).filter(models.User.id == id)
	if not user:
	raise HTTPException(status_code=404)

	user.update(request.dict())
	db.commit()
	return "User updated successfully"

view raw users.fastapi.py hosted with ❤ by GitHub

Now, let's connect the User with their Starship. A pilot can fly more than one starship at different times. They can fly the Falcon one day and the X-Wing the next. This illustrates a one to many relationship between the User and their Starships.

In our code this translates to adding the ForeignKey for User into our Starship and define a relationship to User. In User we will define a relationship with Starship, here back_populates keyword tells SQLAlchemy that there is a relationship on the User side.

	class Starship(Base):
	__tablename__ = 'starship'
	id = Column(String, primary_key=True, index= True)
	name = Column(String)
	price = Column(Float)
	discount = Column(Float)
	user_id = Column(String, ForeignKey('user.id'))
	user = relationship('User', back_populates='starship')
	class User(Base):
	__tablename__ = 'user'
	id = Column(String, primary_key=True, index= True)
	email = Column(String)
	password = Column(String)
	starship = relationship('Starship', back_populates='user')

view raw update.models.fastapi.py hosted with ❤ by GitHub

Routes with APIRouter

Previously, we used app to define our routes directly. This can become cumbersome as we write more code. Instead, we can group related routes together using APIRouter with tags. In addition to that, with APIRouter, we can avoid repeating the base URL for each route, by defining a prefix.
It's as simple as importing APIRouter and defining tags and prefix:

from fastapi import APIRouter

router = APIRouter(tags=["starships"], prefix='/starship')

After you're done, let's do some clean up of our previous code, we will replace the @app decorators with @router. Since you've defined a prefix (/starship) in our router, a route defined as:

@router.post('/')

will be accessible at '/starship' rather than '/'

Authenticating our pilots

Let's create our Sign In endpoint.

First define the new models:

	class SignIn(BaseModel):
	email: str
	password: str

	class Token(BaseModel):
	access_token: str
	token_type: str
	class Config:
	orm_mode = True

	class TokenData(BaseModel):
	email: str

view raw schemas.signin.fastapi.py hosted with ❤ by GitHub

We now need to define two essential functions for our authentication system:

generate_token: This function copies the provided data, adds an expiration timestamp, and encodes everything into a JWT using our secret key and algorithm.

get_auth_user: This function uses oauth2_scheme to extract the token from the request header, decodes it, and verifies the user's email. It can be used as a dependency to protect other routes.

But before we can do any of this, we need a secret key. To generate a secure random hexadecimal string, run:

openssl rand -hex 32

We'll use this generated string as our SECRET_KEY and HS256 as our algorithm.

Below is the complete code for these functions:

	from datetime import datetime, timedelta
	from http.client import HTTPException

	from fastapi import Depends
	from fastapi.security import OAuth2PasswordBearer
	from jose import jwt
	from starship.schemas import TokenData

	SECRET_KEY = SECRET_KEY
	ALGORITHM = "HS256"
	ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 60 * 24 * 7
	oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

	def generate_token(data: dict):
	to_encode = data.copy()
	expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
	to_encode.update({"exp": expire.timestamp()})
	encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
	return encoded_jwt

	def get_auth_user(token: str = Depends(oauth2_scheme)):
	try:
	payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
	email: str = payload.get("sub")
	if email is None:
	raise HTTPException(status_code=401, detail="Could not validate credentials")
	token_data = TokenData(email=email)
	except jwt.ExpiredSignatureError:
	raise HTTPException(status_code=401, detail="Token expired")

view raw token.gen.fastapi.py hosted with ❤ by GitHub

Now onto our Sign In endpoint. What we need to handle is when a user submits their credentials (email and password), the endpoint should be able to:

Query the database to retrieve the user.
Verify the password.

pwd_context.verify()

If both the email and password are correct, generate a JWT token using our generate_token function.
Return the JWT token as an access token along with the token type 'bearer'.

	from fastapi import Depends, APIRouter, HTTPException
	from passlib.context import CryptContext
	from sqlalchemy.orm import Session

	from starship.database import get_db
	from starship.models import User
	from starship.schemas import SignIn
	from starship.key import generate_token

	router = APIRouter(tags=["SignIn"], prefix='/signin')
	pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


	@router.post("/")
	async def signin(request: SignIn, db: Session = Depends(get_db)):
	user = db.query(User).filter(User.email == request.email).first()
	if not user:
	raise HTTPException(status_code=404, detail="User not found")

	if not pwd_context.verify(request.password, user.password):
	raise HTTPException(status_code=404, detail="Incorrect password")

	access_token = generate_token(data={"sub": user.email})
	return {"access_token": access_token, "token_type": "bearer"}

view raw signin.fastapi.py hosted with ❤ by GitHub

Let's try it out:

Awesome! Now that we are able to generate our JWT tokens, we should be able to use it to secure our endpoints. Let's tackle our previous starship endpoint.
Take this operation for instance:

@router.get('/', response_model=List[StarshipResponse])
def get_starships(db: Session = Depends(get_db), auth_user: User = Depends(get_auth_user)):
    return db.query(models.Starship).all()

The get_starships function includes this line now

 auth_user: User = Depends(get_auth_user) parameter.

This dependency ensures that only authenticated users can access the endpoint. If a request does not have valid authentication credentials, FastAPI automatically returns a 401 Unauthorized response.

If you go to your docs you will notice a lock

And If you try it out and hit execute it will return unauthorized error
{ "detail": "Not authenticated" }
Great, that means that your endpoint is now secured. To successfully access this secured endpoint, you need to provide a valid JWT token in the Authorization header of your request. The header should be formatted as follows: