Introduction to CRUD and Why We Need It
Chapter 15: Building a Simple Tasks CRUD
Welcome to Lesson 15! In the previous lesson, we learned how to define data models using Pydantic. Now, we will put that knowledge into action by building a complete CRUD (Create, Read, Update, Delete) API for managing tasks. This is the heart of our project, and by the end of this chapter, you will have a fully functional API that can handle all basic operations on tasks.
What is CRUD and Why Do We Need It?
CRUD stands for Create, Read, Update, and Delete. These are the four basic operations that any persistent storage system should support. In the context of a REST API, CRUD maps directly to HTTP methods:
- Create →
POST(add a new task) - Read →
GET(retrieve one or all tasks) - Update →
PUTorPATCH(modify an existing task) - Delete →
DELETE(remove a task)
Without CRUD, your API would be static. CRUD gives users the power to interact with your data dynamically. For example, a to-do list app must allow users to add new tasks, view them, mark them as complete, and delete them. That is exactly what we are building today.
Setting Up Our In-Memory Data Store
For simplicity, we will store tasks in a Python list. In a real application, you would use a database (which we will cover in Lesson 17). But for now, an in-memory list is perfect for learning the CRUD pattern.
We will also use a simple counter to assign unique IDs to each task. Let's start with our Pydantic model and the data store.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
app = FastAPI()
# Pydantic model for a task
class Task(BaseModel):
id: Optional[int] = None
title: str
description: Optional[str] = None
completed: bool = False
# In-memory storage
tasks_db = []
task_id_counter = 1
Implementing CRUD Operations
1. Create a Task (POST)
To create a task, we accept a Task object (without an ID) and assign it a new ID. Then we append it to our list.
@app.post("/tasks", response_model=Task)
def create_task(task: Task):
global task_id_counter
task.id = task_id_counter
task_id_counter += 1
tasks_db.append(task)
return task
Explanation: The @app.post("/tasks") decorator tells FastAPI to handle POST requests to the /tasks endpoint. The function receives a Task object (validated by Pydantic), assigns it a new ID, adds it to the list, and returns the created task.
2. Read All Tasks (GET)
To retrieve all tasks, we simply return the entire list.
@app.get("/tasks", response_model=list[Task])
def get_tasks():
return tasks_db
Explanation: The @app.get("/tasks") decorator handles GET requests. The response model is a list of Task objects. This endpoint returns all tasks currently in memory.
3. Read a Single Task (GET by ID)
To get a specific task, we need to accept an ID as a path parameter and find the task in our list.
@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int):
for task in tasks_db:
if task.id == task_id:
return task
raise HTTPException(status_code=404, detail="Task not found")
Explanation: The {task_id} in the path is a path parameter. We loop through the list to find the task with the matching ID. If not found, we raise an HTTP 404 error.
4. Update a Task (PUT)
To update a task, we accept the task ID and a new Task object. We find the existing task and update its fields.
@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, updated_task: Task):
for index, task in enumerate(tasks_db):
if task.id == task_id:
# Keep the original ID
updated_task.id = task_id
tasks_db[index] = updated_task
return updated_task
raise HTTPException(status_code=404, detail="Task not found")
Explanation: We use enumerate to get both the index and the task. When we find the matching ID, we replace the old task with the new one (keeping the same ID). If not found, we return a 404 error.
5. Delete a Task (DELETE)
To delete a task, we find it by ID and remove it from the list.
@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
for index, task in enumerate(tasks_db):
if task.id == task_id:
tasks_db.pop(index)
return {"message": "Task deleted successfully"}
raise HTTPException(status_code=404, detail="Task not found")
Explanation: We loop through the list, find the task by ID, and use pop(index) to remove it. We return a success message. If the task is not found, we raise a 404 error.
Complete Code Example
Here is the full code for our CRUD API. Save it as main.py and run it with uvicorn main:app --reload.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
app = FastAPI()
class Task(BaseModel):
id: Optional[int] = None
title: str
description: Optional[str] = None
completed: bool = False
tasks_db = []
task_id_counter = 1
@app.post("/tasks", response_model=Task)
def create_task(task: Task):
global task_id_counter
task.id = task_id_counter
task_id_counter += 1
tasks_db.append(task)
return task
@app.get("/tasks", response_model=list[Task])
def get_tasks():
return tasks_db
@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int):
for task in tasks_db:
if task.id == task_id:
return task
raise HTTPException(status_code=404, detail="Task not found")
@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, updated_task: Task):
for index, task in enumerate(tasks_db):
if task.id == task_id:
updated_task.id = task_id
tasks_db[index] = updated_task
return updated_task
raise HTTPException(status_code=404, detail="Task not found")
@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
for index, task in enumerate(tasks_db):
if task.id == task_id:
tasks_db.pop(index)
return {"message": "Task deleted successfully"}
raise HTTPException(status_code=404, detail="Task not found")
Common Mistakes and How to Avoid Them
- Forgetting to use
globalfor the counter: Inside a function, if you assign a value to a variable that exists outside the function, you must declare it asglobal. Otherwise, Python will create a new local variable. - Not handling missing tasks: Always check if a task exists before trying to update or delete it. Use
HTTPExceptionto return a proper 404 error. - Using the wrong HTTP method: Remember: POST for create, GET for read, PUT for update, DELETE for delete. Using the wrong method will confuse clients.
- Not using
response_model: This ensures FastAPI validates the output and shows the correct schema in the interactive docs.
http://127.0.0.1:8000/docs. It automatically generates a UI for all your endpoints.
Practice Task
Now it's your turn! Extend the CRUD API with the following feature:
- Add a new endpoint
GET /tasks/completedthat returns only the tasks wherecompletedisTrue. - Add a new endpoint
GET /tasks/pendingthat returns only the tasks wherecompletedisFalse.
Hint: Use list comprehension to filter the tasks_db list. For example: [task for task in tasks_db if task.completed].
Once you have implemented these endpoints, test them using the Swagger UI. Congratulations! You have just built a complete CRUD API for tasks. In the next lesson, we will learn how to organize our project into multiple files for better maintainability.

Loading ratings...