Building and Testing FastAPI CRUD APIs with Pytest - A Step-By-Step Guide
Originally developed in the 1940s, APIs have exploded in popularity during the last 30–40 years.
Back in the day, SOAP APIs were popular with Rest and GraphQL becoming the new norm.
As a means to share data between applications and databases, APIs form the building blocks of your software application.
A software application requires the ability to read and write data from a database.
To achieve this, it’s necessary that both the code and the database speak the same language. This is achieved through database drivers.
How useful is software that cannot talk to other software? Or where you cannot decouple the backend from the front end?
API-based microservice architecture has risen in popularity due to the ability to build independent decoupled services that can be maintained, refactored and scaled easily.
Popular API and web-building frameworks for Python include Django, Flask and more recently FastAPI (one I’m liking, a lot).
In this article, you’ll learn how to Build and Test your own API with CRUD functionality (Create, Read, Update and Delete a User) using FastAPI, SQLite and PyTest.
Let’s get started then?
Objectives
By the end of this tutorial you should be able to:
- Develop a REST CRUD API in Python using the FastAPI framework
- Interact with an SQLite database using SQLAlchemy
- Practice Test-Driven Development
- Test your FastAPI API Endpoints/Routes with Pytest and sample payloads
- Handle Errors and Response Formatting
- Document your REST API with Swagger/OpenAPI
Overview Of FastAPI
FastAPI is a modern, fast, and highly performant Python web framework for building APIs with ease.
It was created by Sebastián Ramírez in 2018, and released in 2019.
It is built on top of two other popular Python libraries: Starlette and Pydantic.
Starlette provides the underlying web application framework, while Pydantic is used for data validation and serialization.
FastAPI is designed to be easy to use, with a focus on developer productivity, code quality, and performance.
It also has built-in support for async/await functions, making it much more efficient than the traditional synchronous threading models.
CRUD API
CRUD is an acronym that stands for Create, Read, Update, and Delete.
It’s a set of HTTP methods (POST, GET, PUT/PATCH, DELETE) that are commonly used for performing basic operations on data in a database or other data storage system.
These are widely used in web development for implementing basic functionality in applications such as content management systems, e-commerce platforms, and social networks.
Learning to build robust, highly reliable and scalable CRUD APIs is one of the fundamental requirements of core software engineering and will form the backbone of your backend system.
Project Setup
Let’s get started.
In this project, we’ll create a CRUD API to Create, Read, Update and Delete a User from our Relational Database (using SQLite).
First, start by cloning the repo.
Or, if you prefer to start from scratch feel free to create a folder and call it pytest-fastapi-crud-example or anything you like.
The project has the following structure
The app directory contains the source code, namely
Database
database.py — Create database engine and SQLite settings.1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLITE_DATABASE_URL = "sqlite:///./user.db"
engine = create_engine(
SQLITE_DATABASE_URL, echo=True,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False,
autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Main
main.py — Create the FastAPI client, health checker and middleware.1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27from app import models, user
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
origins = [
"http://localhost:3000",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(user.router, tags=["Users"],
prefix="/api/users")
def root():
return {"message": "The API is LIVE!!"}
As you can see we’ve included a Health Checker to make sure our API is functional when we hit the /api/healthchecker endpoint.
We’ve also included the prefix /api/users so all routes will use this prefix before applying their own specific routing.
Now let’s look at the Schemas and Routes.
Models & Schemas
Database Schema
models.py — Database schema1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16from app.database import Base
from sqlalchemy import TIMESTAMP, Column, String, Boolean
from sqlalchemy.sql import func
from fastapi_utils.guid_type import GUID, GUID_DEFAULT_SQLITE
class User(Base):
__tablename__ = "users"
id = Column(GUID, primary_key=True, default=GUID_DEFAULT_SQLITE)
first_name = Column(String, nullable=False)
last_name = Column(String, nullable=False)
address = Column(String, nullable=True)
activated = Column(Boolean, nullable=False, default=True)
createdAt = Column(
TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
)
updatedAt = Column(TIMESTAMP(timezone=True), default=None, onupdate=func.now())
Our schema is simple, with 7 fields.
The id , activated, createdAt , updatedAt fields are optional with default values specified.
Ideally, in production, you do NOT want the user to set these unless there’s a good reason.
Schemas (API Schema)
Ideally, we want to define all schemas using OpenAPI or JSON schema and generate the code bindings using a data model code generator.
In this way, the schemas can be ported and code bindings generated for any programming language.
But as this is a FastAPI tutorial let’s keep the discussion about Python.
schemas.py — User Base Schema and Response Schema1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22from datetime import datetime
from typing import List
from pydantic import BaseModel
class UserBaseSchema(BaseModel):
id: str | None = None
first_name: str
last_name: str
address: str | None = None
activated: bool = False
createdAt: datetime | None = None
updatedAt: datetime | None = None
class Config:
orm_mode = True
allow_population_by_field_name = True
arbitrary_types_allowed = True
class ListUserResponse(BaseModel):
status: str
results: int
users: List[UserBaseSchema]
This file contains the base schema for an API request (UserBaseSchema) defined using Pydantic.
In this example, it is similar to the database model above, but it doesn’t have to be by any means.
This file also contains the schema for the API response.
Routes
Moving on to the API routes, we have a user.py file that contains the routes for our CRUD API.
user.py — API Routes and response formatting1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72import app.schemas as schemas, app.models as models
from sqlalchemy.orm import Session
from fastapi import Depends, HTTPException, status, APIRouter
from app.database import get_db
router = APIRouter()
def get_users(
db: Session = Depends(get_db), limit: int = 10, page: int = 1, search: str = ""
):
skip = (page - 1) * limit
users = (
db.query(models.User)
.filter(models.User.first_name.contains(search))
.limit(limit)
.offset(skip)
.all()
)
return {"Status": "Success", "Results": len(users), "Users": users}
def create_user(payload: schemas.UserBaseSchema, db: Session = Depends(get_db)):
new_user = models.User(**payload.dict())
db.add(new_user)
db.commit()
db.refresh(new_user)
return {"Status": "Success", "User": new_user}
def update_user(
userId: str, payload: schemas.UserBaseSchema, db: Session = Depends(get_db)
):
user_query = db.query(models.User).filter(models.User.id == userId)
db_user = user_query.first()
if not db_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"No User with this id: {userId} found",
)
update_data = payload.dict(exclude_unset=True)
user_query.filter(models.User.id == userId).update(
update_data, synchronize_session=False
)
db.commit()
db.refresh(db_user)
return {"Status": "Success", "User": db_user}
def get_user(userId: str, db: Session = Depends(get_db)):
user = db.query(models.User).filter(models.User.id == userId).first()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"No User with this id: `{userId}` found",
)
return {"Status": "Success", "User": user}
def delete_user(userId: str, db: Session = Depends(get_db)):
user_query = db.query(models.User).filter(models.User.id == userId)
user = user_query.first()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"No user with this id: {userId} found",
)
user_query.delete(synchronize_session=False)
db.commit()
return {"Status": "Success", "Message": "User deleted successfully"}
This is one of the most important files for our API where we specify what happens when each route is called e.g. POST , GET , PATCH and DELETE .
We will break down each of these routes separately below.
Create User (POST)
1 |
|
Our Create User POSTroute creates a new record in the database when the /api/users endpoint is called with a POST request.
It takes the JSON payload1
2
3
4
5
6{
"first_name": "Eric",
"last_name": "Sales De Andrade",
"address": "112 Fake Street, Fakeville",
"activated": true
}
As we saw before we can override the id, created, updated dates etc in this example.
But in a production setting, we wouldn’t do this as it could mess up our data and cause all sorts of errors.
It returns a 201 HTTP response code on successful record creation and 422 in case of client payload or schema validation errors.
If you’re unfamiliar with HTTP response codes, here’s a good start.
Read User (GET)
1 |
|
Our Read User GET route reads an individual user from the database based on the userId provided in the endpoint.
A successful response returns a 200 status code and a 404 if the user is not found.
Read User/s (GET)
1 |
|
A popular route alongside get user is the get_users route to return ALL users from the database.
Update User (PATCH)
1 |
|
More often than not, our clients may want to update their information.
Hence we use the Update User PATCH route which allows the API client to update their data in our database.
Delete User (Delete)
1 |
|
With data privacy laws like GDPR, a delete option is absolutely necessary.
From a user perspective, it’s useful to be able to delete data too.
We do this using our Delete User DELETE route, where we can specify the userId to be deleted as part of the delete request.
Testing the API using PyTest
OK, now that we’ve implemented the CRUD API, let’s look at how to test it.
Luckily for us, the creator of FastAPI has already thought of it and made it fairly easy.
Here’s a note from the docs.
Thanks to Starlette, testing FastAPI applications is easy and enjoyable.
It is based on HTTPX, which in turn is designed based on Requests, so it’s very familiar and intuitive.
As you predicted, we’ll be using PyTest :)
Setting Up A TestClient
You can set up a TestClient for FastAPI using1
2
3
4from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
This client will then be used to call our API routes.
Don’t forget to import the app from the app/mainfile so your test client actually uses that one instead of defining a new app.
Testing Each Endpoint Route
Now let’s test all our API routes. We want to make sure it all works and return the correct responses.
Note — The below Unit Tests WILL create an actual record in the database and delete it.
We’ll discuss mocking/patching later.
- Test
healthchecker:
1 | def test_root(): |
This test calls the /api/healthchecker endpoint and asserts if the expected message is returned.
2. Test create_user :1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23def test_create_user():
sample_payload = {
"id": user_id,
"first_name": "PLACEHOLDER",
"last_name": "PLACEHOLDER",
"address": "PLACEHOLDER",
"activated": False,
"createdAt": "2023-03-17T00:04:32",
}
response = client.post("/api/users/", json=sample_payload)
assert response.status_code == 201
assert response.json() == {
"Status": "Success",
"User": {
"first_name": "PLACEHOLDER",
"last_name": "PLACEHOLDER",
"activated": False,
"createdAt": "2023-03-17T00:04:32",
"id": user_id,
"address": "PLACEHOLDER",
"updatedAt": None,
},
}
Here we prepare a sample payload and pass it to the TestClient.
This then returns an HTTP status code and created record.
Note that we’ve passed all parameters here (e.g. id , createdAt , updatedAt ) as we need to assert against expected values.
- Test get_user:
1 | def test_get_user(): |
This test gets the created user from the database.
- Test update_user :
1 | def test_update_user(): |
Here we update the PLACEHOLDER values.
- Test
delete_user:
Lastly, to keep with best practices and delete our used resources, we delete the test user using the DELETE route.1
2
3
4
5
6
7def test_delete_user():
response = client.delete(f"/api/users/{user_id}")
assert response.status_code == 200
assert response.json() == {
"Status": "Success",
"Message": "User deleted successfully",
}
- Test
get_user_not_found:
Now let’s quickly test a simple edge case i.e. User Not Found.
It’s important to give our client the correct message and status code in this case.1
2
3
4
5
6
7
8def test_get_user_not_found():
response = client.get(
f"/api/users/16303002-876a-4f39-ad16-e715f151bab3"
) # GUID not in DB
assert response.status_code == 404
assert response.json() == {
"detail": "No User with this id: `16303002-876a-4f39-ad16-e715f151bab3` found"
}
As expected, our API returns a 404 error code with a helpful error message saying a user with that ID was not found.
Running the Unit Tests
To run the unit tests, simply run1
pytest tests/unit/test_crud_api.py -v -s
Mocking or Real Testing?
Perhaps you’re wondering if you should you mock or write to a real database in practice.
We’ve covered mocking using Monkeypath and in our article on Testing REST APIs.
The truth is (like all things in tech) — there is no one single right answer. It all depends on
- The development time
- Use case
- Risk and Tolerance for failure
- Data usage and retention policy
- Delivery deadlines
- Type of data handled (PII, PCI etc.)
- Permissions and Security
The good news is you can always use a hybrid approach.
Perhaps testing against a real database in Dev/Staging and mock in Prod. Or the other way around.
Ideally, you want to test the entire E2E functionality including the database drivers (read and write to the DB) to ensure permissions and roles are correctly set up.
Testing Using Postman
Now that our source code and unit tests are written, it’s time to test each of our API endpoints using Postman (or another API client tool).
If you’re unfamiliar with Postman, you can check out their Getting Started Guide here.
To start the API server, run1
uvicorn app.main:app --host localhost --port 8000 --reload
You should be able to see the API server started and generating logs, like this
Fire up the API Server
create_user
Create a user
read_user
Read user
read_users
Read users
update_user
Read user before updating
Update user
delete_user
Delete user
Error Handling and Response Formatting
As an API developer, one of the most helpful things you can do for your clients is produce easy to understand documentation.
This is simply information on how to use the API and give the client useful error messages with the correct HTTP response codes.
- Informational responses (100–199)
- Successful responses (200–299)
- Redirection messages (300–399)
- Client error responses (400–499)
- Server error responses (500–599)
You can go into very specific error codes, but keep these ranges in mind, test and include error codes and correct responses for variations of client input.
Documenting Your Rest API with Swagger and OpenAPI
What good is an API if others (consumers) don’t know how to use it?
Documentation is very important, and personally, I’m a big fan of the FastAPI documentation itself.
The bare necessary documentation you should provide is a Swagger and a list of endpoints with acceptable payloads and response formats.
Loosely defined, Swagger is
“Swagger is a set of rules, specifications and tools that help us document our APIs.”
This includes information about the endpoints, how to use them, status codes, response formatting, payloads etc.
Lucky for us, the creator of FastAPI has already thought this through.
It has an inbuilt Swagger UI to test your API should you choose not to use Postman.
Once you run the API server, you can access the inbuilt Swagger UI on [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs) or whatever port you started it on.
You can then go about testing various endpoints.
Conclusion
In conclusion, building and testing a FastAPI CRUD API with Pytest is a powerful way to create a reliable and efficient API.
In this article, you went through creating your own CRUD API to create, read, update and delete a user using FastAPI and SQLite (via SQLAlchemy)
You learnt how to define database and model schemas, write routes with HTTP status codes and handle responses.
Lastly and most importantly you learnt how to test your CRUD API which can form a powerful part of your CI pipeline to ensure everything works as expected.
With FastAPI’s intuitive syntax and Pytest’s testing capabilities, you can quickly and easily build scalable, robust and reliable APIs.
If you have ideas for improvement or like for me to cover anything specific, please send me a message via Twitter, GitHub or Email.
Till the next time… Cheers!