Build a Simple Tasks CRUD API with FastAPI

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 → PUT or PATCH (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 global for the counter: Inside a function, if you assign a value to a variable that exists outside the function, you must declare it as global. 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 HTTPException to 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.

Practice Task

Now it's your turn! Extend the CRUD API with the following feature:

• Add a new endpoint GET /tasks/completed that returns only the tasks where completed is True.
• Add a new endpoint GET /tasks/pending that returns only the tasks where completed is False.

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.