Quick guide to implementing CRUD operations with FastAPI
In this post, we will explain how to implement CRUD operations with FastAPI from something basic to something more complete like using a database.
Introduction
The recent rise in popularity of Python as a programming language is mainly due to its libraries used in the field of data science applications. However, Python is also widely used for web application development, thanks to the abundance of its web application frameworks.
FastAPI is the latest entrant in the long line of Python web application frameworks. However, it is not just another framework as it has some distinctive advantages over the others. Considered one of the “fastest,” FastAPI leverages the capabilities of modern Python.
What is CRUD?
CRUD operations are essential in any web application that involves data storage and retrieval. These operations allow users to create new records, retrieve existing records, update existing records, and delete records from a database.
Here is a detailed explanation of CRUD operations:
- Create ©: The CREATE operation does what its name suggests. It means creating an entry. This entry can be an account, user information, an entry, or a task.
- Read (R): READ operation means accessing inputs or inputs of the user interface. That is, see them. Again, the input can be anything from user information to social media posts, among others.
- Update (U): UPDATE is the operation that allows modifying existing data, that is, editing the data.
- Delete (D): DELETE is to delete an entry from the user interface and the database.
Objective of the project
Build a CRUD API using FastAPI and Python, allowing users to manage a collection of items (for example, books, products, or users). This project will teach students how to create, read, update, and delete data using a RESTful API design.
Previous requirements
To get the most out of this tutorial, you should understand:
- Basic knowledge of Python.
- Knowledge of RESTful APIs and HTTP methods.
- IDE (such as Visual Studio Code) and Python 3.6+ installed.
- Know concepts such as Python Typing de python y and FastAPI Response Model.
- (Optional) What is SQLAlchemy / Engine — MongoDB
Used technology
- FastAPI
- Uvicorn (to run FastAPI applications on a remote server)
- (Optional) PostgreSQL (or any other relational or non-relational database)
Step by step guide
To implement CRUD functionality with FastAPI, follow these steps:
Project settings
- Create a new Python project directory.
- Set up a virtual environment
- Install FastAPI and Uvicorn
Let’s verify that we are within the virtual environment
Let’s verify that Python is working
Then we run the following command on the command line to install FastAPI
pip install fastapi uvicorn
Create a basic FastAPI application
- Create a new file (e.g. `main.py`)
- Import FastAPI and create a basic instance
from fastapi import FastAPIapp = FastAPI()@app.get('/')
def welcome():
return {'message': 'Welcome to my FastAPI application'}
Define the data model
- Create a Python class to define the data model (e.g. a `Book` class with title, author, etc.)
- Defines Pydantic models for input validation
from pydantic import BaseModelclass Article(BaseModel):
id: int
name: str
price: floatCreate routes and CRUD handlers
This code in FastAPI establishes the basic CRUD operations for a list of items. Allows reading the entire list, creating new items, updating based on ID, and deleting items by ID. Simplifies article collection management in a web application.
articles = []@app.get("/articles", response_model=List[Article])
async def read_articles():
return articles@app.post("/articles", response_model=Article)
async def create_article(article: Article):
articles.append(article)
return article@app.put("/articles/{article_id}", response_model=Article)
async def update_article(article_id: int, article: Article):
articles[article_id] = article
return article@app.delete("/articles/{article_id}")
async def delete_article(article_id: int):
del articles[article_id]
return {"message": "Article deleted"}
Running the application
uvicorn main:app --reloadThis will start the FastAPI application on the default port (typically 8000) with auto-reload enabled, so the application will automatically reload when you make changes to the code.
Then the complete code would look like this:
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Listapp = FastAPI()# Definition of a Pydantic model named "Item" with three fields: id, name, and price.
class Article(BaseModel):
id: int
name: str
price: float# Empty list to store created articles.
articles = []# Route for the home page that returns a welcome message.
@app.get('/')
def welcome():
return {'message': 'Welcome to my FastAPI application'}# Route to get all articles stored in the list.
# The "response_model" parameter specifies that the response will be a list of "Article" objects.
@app.get("/articles", response_model=List[Article])
async def read_articles():
return articles# Route to create a new article.
# The "response_model" parameter specifies that the response will be an "Article" object.
@app.post("/articles", response_model=Article)
async def create_article(article: Article):
articles.append(article) # Adds the article to the list.
return article# Route to update an existing article by its ID.
# The "response_model" parameter specifies that the response will be an "Article" object.
@app.put("/articles/{article_id}", response_model=Article)
async def update_article(article_id: int, article: Article):
articles[article_id] = article # Updates the article in the list.
return article# Route to delete an article by its ID.
# "response_model" is not specified since no object is returned in the response.
@app.delete("/articles/{article_id}")
async def delete_article(article_id: int):
del articles[article_id] # Deletes the article from the list.
return {"message": "Article deleted"} # Returns an informative message.
Let’s remember that FastAPI creates an interactive and automatically created documentation on our api, so to view it we put the following address “http://localhost:8000/docs”
Let’s try CRUD operations
It’s time to put our basic API to the test.
Use an HTTP client tool (for example, Postman or Thunder Client) to send requests and test the Create, Read, Update, and Delete capabilities.
These are some examples of requests:
POST http://localhost:8000/items
{
"id": 1,
"name": "Apple",
"price": 0.5
}Send a GET request to retrieve all items
GET http://localhost:8000/itemsSends a PUT request with the request body to update an item.
PUT http://localhost:8000/items/1
{
"id": 1,
"name": "Orange",
"price": 1.0
}Keep in mind that this is a simple example, in real applications more programming logic is needed so that there is no problem with the routes, in our example we use “…/articles/0” because remember that the lists start with index zero
Send a DELETE request to delete an item
Since we eliminate the only element that was there, we are left with an empty list
With this simple example you should have already understood that the FastAPI application consists of several operational functions. Each operation function is invoked by the decorator of the corresponding HTTP method to which it is assigned. The HTTP POST, GET, PUT, and DELETE methods respectively create a resource, retrieve one or more resources available on the server, and update or delete one or more resources.
To perform persistent CRUD operations, the application needs to interact with a data storage and retrieval system. Applications generally use relational databases as the back-end.
As this was an introductory and quite practical blog, we are not going to go too deep into seeing the connection with a real database, so here I leave the code so you can get an idea of what it would be like to use FastAPI with SQLite
main.py: Contains FastAPI application configuration and CRUD routes.
# crud/main.py
from typing import List
from fastapi import FastAPI, status, HTTPException, Depends
from database import Base, engine, SessionLocal
from sqlalchemy.orm import Session
import models
import schemasBase.metadata.create_all(engine) # Create the database# Initialize the application
app = FastAPI()# Helper function to get the database session
def get_session():
session = SessionLocal()
try:
yield session
finally:
session.close()@app.get("/")
def home():
return "all"@app.post("/todo", response_model=schemas.Task, status_code=status.HTTP_201_CREATED)
def create_task(task: schemas.TaskCreate, session: Session = Depends(get_session)):
task_db = models.Task(task=task.task) session.add(task_db)
session.commit()
session.refresh(task_db) return task_db@app.get("/todo/{id}", response_model=schemas.Task)
def read_task(id: int, session: Session = Depends(get_session)):
task = session.query(models.Task).get(id) # Get item with given id # Check if id exists. If not, return 404 not found response
if not task:
raise HTTPException(status_code=404, detail=f"Task with id {id} not found") return task@app.put("/todo/{id}", response_model=schemas.Task)
def update_task(id: int, task: str, session: Session = Depends(get_session)):
task_db = session.query(models.Task).get(id) # Get given id if task_db:
task_db.task = task
session.commit() # Check if id exists. If not, return 404 not found response
if not task_db:
raise HTTPException(status_code=404, detail=f"Task with id {id} not found") return task_db@app.delete("/todo/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(id: int, session: Session = Depends(get_session)):
# Get given id
task_db = session.query(models.Task).get(id) # If task with given id exists, delete it from the database. Otherwise, raise 404 error
if task_db:
session.delete(task_db)
session.commit()
else:
raise HTTPException(status_code=404, detail=f"Task with id {id} not found") return None@app.get("/todo", response_model=List[schemas.Task])
def read_task_list(session: Session = Depends(get_session)):
task_list = session.query(models.Task).all() # Get all tasks return task_list
database.py: SQLAlchemy configuration, including creating an SQLite database engine and creating a session.
# fastAPI\crud_todo\database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker# Create an instance of SQLite engine
engine = create_engine("sqlite:///fastapidb.db")# Create an instance of DeclarativeMeta
Base = declarative_base()# Create the SessionLocal class from sessionmaker factory
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False
models.py: Define the structure of the tasks table using SQLAlchemy.
# fastAPI\crud_todo\models.py
from sqlalchemy import Column, Integer, String
from database import Base# Define the Task class from Base
class Task(Base):
__tablename__ = 'tasks'
id = Column(Integer, primary_key=True)
task = Column(String(256))schemas.py: Defines the Pydantic models used to validate input and output data.
# fastAPI\crud_todo\schemas.py
from pydantic import BaseModel# Create Task schema (Pydantic Model)
class TaskCreate(BaseModel):
task: str# Full Task schema (Pydantic Model)
class Task(BaseModel):
id: int
task: str class Config:
orm_mode = True
Conclusions
This project guide provides hands-on experience in creating CRUD APIs using FastAPI.
It covers core aspects of API development, including defining data models, processing requests, and interacting with databases.
This is a hands-on exercise for new programmers who want to strengthen their knowledge of Python and API development.
Next steps
In the next steps, consider strengthening your FastAPI application by incorporating advanced features such as authentication and authorization to protect CRUD operations. You can also improve the user experience by implementing paging and exploring integrations with more complex databases or cloud services. Additionally, consider developing a UI with React or another front-end framework to provide a friendlier experience when interacting with the API.
