In this post, we will see how to create a web API to manage all CRUD operations for a class called User, using FastAPI.
But first of all, what is FastAPI?
From the official web site:
FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.8+ based on standard Python type hints.
The key features are:
- Fast: Very high performance, on par with NodeJS and Go (thanks to Starlette and Pydantic).
- Fast to code: Increase the speed to develop features by about 200% to 300%. *
- Fewer bugs: Reduce about 40% of human (developer) induced errors. *
- Intuitive: Great editor support. Completion everywhere. Less time debugging.
- Easy: Designed to be easy to use and learn. Less time reading docs.
- Short: Minimize code duplication. Multiple features from each parameter declaration. Fewer bugs.
- Robust: Get production-ready code. With automatic interactive documentation.
- Standards-based: Based on (and fully compatible with) the open standards for APIs: OpenAPI (previously known as Swagger) and JSON Schema.
We start defining the class User and a class called Core that, we will use as our Business Layer:
[USER.PY]
import uuid
class User:
def __init__(self, input_username, input_password, input_id=None):
"""
Initializes a User object.
Args:
input_username (str): The user's chosen username.
input_password (str): The user's password.
input_id (str, optional): An optional pre-existing ID. Defaults to None.
"""
self.username = input_username
self.password = input_password
# Generate a new UUID if no input_id was given
self.id = uuid.uuid4() if input_id == None else input_id
[CORE.PY]
import asyncio
from User import User
class Core:
def __init__(self):
# Initialize an instance variable to store user list.
# It is marked as private with double underscores to indicate
# that it should not be accessed directly from outside the class.
self.__lst_user = []
async def _create_default_users(self):
# This generates a list of default users asynchronously.
# The method is intended for internal use.
users = []
for index in range(1, 6):
# The await keyword is used to simulate an asynchronous operation like a database call.
await asyncio.sleep(0.1)
# A new User object is created and appended to the users list.
users.append(User(f"Username{index}", f"Pass12{index}"))
# After the loop completes, the list of user objects is returned.
return users
async def get_all_users(self):
# This returns all users.
# If no users have been created yet, it first generates the default users.
if len(self.__lst_user) == 0:
# Await the creation of default users if the list is empty.
self.__lst_user = await self._create_default_users()
# Return the list of users, which is now populated with default users if it was previously empty.
return self.__lst_user
Then, we install FastAPI and an ASGI server like Uvicorn, using the command:
py.exe -m pip install fastapi uvicorn
Before continuing, I would like to explain what ASGI is:
ASGI stands for Asynchronous Server Gateway Interface and it’s a specification for Python web servers and applications to communicate with each other, and it’s particularly designed to support asynchronous programming.
Finally, we create the main.py file where we will add all endpoints for our API service:
[MAIN.PY]
from fastapi import FastAPI
from Core import Core
app = FastAPI()
core = Core()
@app.get("/users")
async def get_users():
return await core.get_all_users()
Now, we will run the application and then, using a API client, we will check that everything works fine.
To run the application, we use the command:
py.exe -m uvicorn main:app --reload
Before to add all CRUD operations, I want to talk about two important things.
The first is to define a Pydantic model to use as input to the endpoints.
It offers several significant benefits, especially when dealing with input validation, serialization, and documentation in web APIs.
In this case, we will define a Pydantic class by adding the check to the password which must be at least 6 characters long.
[USERBASEMODEL.PY]
from pydantic import BaseModel, validator
class User(BaseModel):
username: str
password: str
@validator('password')
def password_must_be_long(cls, value):
if len(value) < 6:
raise ValueError('Password must be at least 6 characters long')
return value
The second is to add a basic authentication using FastAPI’s security utilities. A common approach is to use HTTP Basic Authentication where, the client sends a username and password with each request, and the server validates these credentials.
To achieve it, we add a method called check_credential in the main.py:
security = HTTPBasic()
def check_credentials(credentials: HTTPBasicCredentials = Depends(security)):
# Obviously, this check credential workflow, is ONLY for a demo and not for a real project!
correct_username = secrets.compare_digest(credentials.username, "admin")
correct_password = secrets.compare_digest(credentials.password, "pass123")
if not (correct_username and correct_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Basic"},
)
return credentials
Now, before to create all endpoints in the main file, I prefer adding all methods in our Business Layer:
[CORE.PY]
import asyncio
from User import User
class Core:
def __init__(self):
# Initialize an instance variable to store user list.
# It is marked as private with double underscores to indicate
# that it should not be accessed directly from outside the class.
self.__lst_user = []
async def _create_default_users(self):
# This generates a list of default users asynchronously.
# The method is intended for internal use.
users = []
for index in range(1, 6):
# The await keyword is used to simulate an asynchronous operation like a database call.
await asyncio.sleep(0.1)
# A new User object is created and appended to the users list.
users.append(User(f"Username{index}", f"Pass12{index}"))
# After the loop completes, the list of user objects is returned.
return users
async def get_all_users(self):
# This returns all users.
# If no users have been created yet, it first generates the default users.
if len(self.__lst_user) == 0:
# Await the creation of default users if the list is empty.
self.__lst_user = await self._create_default_users()
# Return the list of users, which is now populated with default users if it was previously empty.
return self.__lst_user
async def insert_user(self, username, password):
try:
# Create a new User object with the provided username and password.
obj_user = User(username, password)
# Append the newly created user object to the list of users.
self.__lst_user.append(obj_user)
# Simulate an asynchronous operation, such as a database insert.
await asyncio.sleep(0.2)
except Exception as error:
# If any error occurs during the process, print the error.
print(error)
# Re-raise the error to be handled or logged by the caller of this method.
raise
async def get_user_by_id(self, id):
# Use the filter function with a lambda expression to filter users by ID.
filtered_users = filter(lambda user: str(user.id) == id, self.__lst_user)
# Simulate an asynchronous operation, such as fetching a user from a database.
await asyncio.sleep(0.1)
# Use the next function to get the first user from the filtered_users iterator.
# If no users match the filter (i.e., the iterator is empty), return None.
return next(filtered_users, None)
async def update_user(self, id, password):
try:
# Attempt to find the user by ID using the get_user_by_id method.
user = await self.get_user_by_id(id)
# Check if the user could not be found. If not, return False indicating failure to update.
if user == None:
return False
# If the user is found, update the user's password with the new password provided.
user.password = password
# Simulate an asynchronous operation, such as updating a user record in a database.
await asyncio.sleep(0.2)
# Return True to indicate that the user's password was successfully updated.
return True
except Exception as error:
# If any exceptions occur during the process, print the error to the console.
print(error)
# Re-raise the error to be handled or logged by the caller of this method.
raise
async def delete_user(self, id):
try:
# Attempt to find the user by their ID.
user = await self.get_user_by_id(id)
# If no user with the given ID was found, return False to indicate failure to delete.
if user == None:
return False
# If the user is found, remove them from the list of users.
self.__lst_user.remove(user)
# Simulate an asynchronous operation, such as deleting a user record from a database.
await asyncio.sleep(0.2)
# Return True to indicate that the user was successfully deleted.
return True
except Exception as error:
# If any exceptions occur during the process, print the error to the console.
# This could be useful for debugging or logging error details.
print(error)
# Re-raise the error to allow it to be handled or logged by the caller of this method.
raise
Finally, we will add all the endpoints in the main file step by step:
INSERT USER
@app.post("/user", status_code=status.HTTP_201_CREATED)
async def create_user(user: User, credentials: HTTPBasicCredentials = Depends(check_credentials)):
# async def create_user(user_in: User):
try:
await core.insert_user(user.username, user.password)
return {"message": "User created successfully"}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
SELECT USER BY ID
@app.get("/user/{id}")
async def get_user_by_id(id: str):
user = await core.get_user_by_id(id)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
UPDATE USER
@app.put("/user/{id}", status_code=status.HTTP_200_OK)
async def update_user(id: str, user: User):
try:
result = await core.update_user(id, user.password)
if result:
return {"message": "User updated successfully"}
else:
raise HTTPException(status_code=404, detail="User not found")
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
DELETE USER
@app.delete("/user/{id}", status_code=status.HTTP_200_OK)
async def delete_user(id: str):
try:
result = await core.delete_user(id)
if result:
return {"message": "User deleted successfully"}
else:
raise HTTPException(status_code=404, detail="User not found")
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
[MAIN.PY]
from fastapi import FastAPI, HTTPException, status, Depends
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from Core import Core
from UserBaseModel import User
import secrets
app = FastAPI()
core = Core()
security = HTTPBasic()
def check_credentials(credentials: HTTPBasicCredentials = Depends(security)):
# Obviously, this check credential workflow, is ONLY for a demo and not for a real project!
correct_username = secrets.compare_digest(credentials.username, "admin")
correct_password = secrets.compare_digest(credentials.password, "pass123")
if not (correct_username and correct_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Basic"},
)
return credentials
@app.get("/users")
async def get_users():
return await core.get_all_users()
@app.post("/user", status_code=status.HTTP_201_CREATED)
async def create_user(user: User, credentials: HTTPBasicCredentials = Depends(check_credentials)):
# async def create_user(user_in: User):
try:
await core.insert_user(user.username, user.password)
return {"message": "User created successfully"}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@app.get("/user/{id}")
async def get_user_by_id(id: str):
user = await core.get_user_by_id(id)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.put("/user/{id}", status_code=status.HTTP_200_OK)
async def update_user(id: str, user: User):
try:
result = await core.update_user(id, user.password)
if result:
return {"message": "User updated successfully"}
else:
raise HTTPException(status_code=404, detail="User not found")
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@app.delete("/user/{id}", status_code=status.HTTP_200_OK)
async def delete_user(id: str):
try:
result = await core.delete_user(id)
if result:
return {"message": "User deleted successfully"}
else:
raise HTTPException(status_code=404, detail="User not found")
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))