KAHIBARO
Discord Login Register ☰
☰

29.4. CRUD Operations

Table of Contents

Understanding CRUD in the Task Management API

In this chapter you will implement the four basic operations of almost every backend application: Create, Read, Update, Delete (CRUD) for tasks in your Task Management API.

We will assume you already have:

Here we connect those pieces into real, usable endpoints.

CRUD stands for:

  • Create
  • Read
  • Update
  • Delete
    Every resource-based API will implement these in some form.

For examples, we will use:

You can adapt to your own stack if it differs.

Typical Task fields we will use:

FieldTypeDescription
idintUnique task identifier (primary key)
titlestringShort task title
descriptionstring/nullOptional detailed description
is_doneboolCompletion flag
created_atdatetimeWhen the task was created
updated_atdatetimeLast update time

Designing Request and Response Models

Before implementing CRUD operations, define Pydantic models for input and output. These models make your API clear and consistent.

A common pattern uses three models:

Example:

python 
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
class TaskBase(BaseModel):
    title: str = Field(..., max_length=200)
    description: Optional[str] = None
    is_done: bool = False
class TaskCreate(TaskBase):
    # All fields required for creation already in TaskBase
    pass
class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, max_length=200)
    description: Optional[str] = None
    is_done: Optional[bool] = None
class TaskRead(TaskBase):
    id: int
    created_at: datetime
    updated_at: datetime
    class Config:
        orm_mode = True

Some points:

Mapping CRUD to HTTP Methods and URLs

A simple, REST style design for tasks:

OperationHTTP MethodURL patternDescription
CreatePOST/tasksCreate a new task
Read (list)GET/tasksList all tasks
Read (one)GET/tasks/{task_id}Get a single task by id
UpdatePUT or PATCH/tasks/{task_id}Update an existing task
DeleteDELETE/tasks/{task_id}Delete a task

Important design rule

  • Use collection URL (/tasks) for actions on many resources like create and list.
  • Use item URL (/tasks/{task_id}) for actions on a single resource like get, update, delete.

Next we will implement each of these operations.

Create: Adding a New Task

The Create operation usually uses POST /tasks.
The client sends a JSON body with data for the new task, and the server returns the created task, often with HTTP status 201 Created.

Example request body:

json 
{
  "title": "Finish course chapter",
  "description": "Write CRUD section for Task API",
  "is_done": false
}

Example SQLAlchemy model

We will assume something like this:

python 
from datetime import datetime
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from .database import Base  # Your SQLAlchemy Base
class Task(Base):
    __tablename__ = "tasks"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    description = Column(String, nullable=True)
    is_done = Column(Boolean, default=False, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, nullable=False)

POST /tasks endpoint

python 
from fastapi import APIRouter, Depends, status
from sqlalchemy.orm import Session
from . import models, schemas
from .database import get_db
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.post(
    "",
    response_model=schemas.TaskRead,
    status_code=status.HTTP_201_CREATED,
)
def create_task(task_in: schemas.TaskCreate, db: Session = Depends(get_db)):
    task = models.Task(
        title=task_in.title,
        description=task_in.description,
        is_done=task_in.is_done,
    )
    db.add(task)
    db.commit()
    db.refresh(task)  # Load generated id and timestamps from DB
    return task

What happens:

  1. FastAPI parses the JSON body into TaskCreate.
  2. You create a Task ORM object with that data.
  3. You add and commit it to the database.
  4. db.refresh loads DB-generated values such as id and timestamps.
  5. The task is returned as TaskRead due to response_model=schemas.TaskRead.

Common mistakes to avoid:

Suggested behavior:

Read: Listing and Fetching Tasks

The Read operations are:

GET /tasks list all tasks

Basic version:

python 
from typing import List
from fastapi import Query
@router.get(
    "",
    response_model=List[schemas.TaskRead],
)
def list_tasks(
    db: Session = Depends(get_db),
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
):
    tasks = db.query(models.Task).offset(skip).limit(limit).all()
    return tasks

Notes:

Example response:

json 
[
  {
    "id": 1,
    "title": "Finish course chapter",
    "description": "Write CRUD section for Task API",
    "is_done": false,
    "created_at": "2026-08-28T12:00:00Z",
    "updated_at": "2026-08-28T12:00:00Z"
  },
  {
    "id": 2,
    "title": "Review pull request",
    "description": null,
    "is_done": true,
    "created_at": "2026-08-28T14:00:00Z",
    "updated_at": "2026-08-28T14:15:00Z"
  }
]

You can later extend this with filtering, searching, or sorting.

GET /tasks/{task_id} get a single task

python 
from fastapi import HTTPException
@router.get(
    "/{task_id}",
    response_model=schemas.TaskRead,
)
def get_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    return task

Important points:

Important behavior rule for Read operations

  • When a resource with a given id does not exist, return HTTP 404.
    Do not silently return null or an empty object.

Update: Modifying Existing Tasks

For updates you typically use:

In many simple APIs you can implement only one of them, usually something closer to partial update using PATCH. Here we focus on a partial update with TaskUpdate.

Deciding between PUT and PATCH

To avoid confusion, we will implement:

PATCH /tasks/{task_id}

python 
from fastapi import Body
@router.patch(
    "/{task_id}",
    response_model=schemas.TaskRead,
)
def update_task(
    task_id: int,
    task_in: schemas.TaskUpdate = Body(...),
    db: Session = Depends(get_db),
):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    # Only update fields that were provided
    update_data = task_in.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(task, key, value)
    db.commit()
    db.refresh(task)
    return task

Explanation:

Example patch request:

http 
PATCH /tasks/1
Content-Type: application/json
{
  "is_done": true
}

Example response:

json 
{
  "id": 1,
  "title": "Finish course chapter",
  "description": "Write CRUD section for Task API",
  "is_done": true,
  "created_at": "2026-08-28T12:00:00Z",
  "updated_at": "2026-08-28T15:00:00Z"
}

Optional: PUT for full update

If you want to support PUT:

Simple version using TaskCreate as body:

python 
@router.put(
    "/{task_id}",
    response_model=schemas.TaskRead,
)
def replace_task(
    task_id: int,
    task_in: schemas.TaskCreate,
    db: Session = Depends(get_db),
):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    task.title = task_in.title
    task.description = task_in.description
    task.is_done = task_in.is_done
    db.commit()
    db.refresh(task)
    return task

Update behavior rules

  • If the resource does not exist, return 404 Not Found.
  • Do not let the client change server-managed fields such as id or created_at.
  • Use PATCH for partial updates, and be consistent about it.

Delete: Removing Tasks

The Delete operation uses DELETE /tasks/{task_id}.
The server removes the task and usually returns:

We will use 204 No Content, which is common and simple.

DELETE /tasks/{task_id}

python 
@router.delete(
    "/{task_id}",
    status_code=status.HTTP_204_NO_CONTENT,
)
def delete_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    db.delete(task)
    db.commit()
    # No return body for 204

Behavior summary:

Alternative: soft deletes

In some applications you do not physically delete records, you mark them as deleted with a flag like is_deleted. For this project you likely do a real delete, but know soft delete is a common alternative.

Example soft delete code:

python 
def delete_task_soft(task_id: int, db: Session):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    task.is_deleted = True
    db.commit()

Typical CRUD Workflow in the Task API

To see CRUD in action, imagine the following sequence for a single task.

  1. Create a task
http 
   POST /tasks
   Content-Type: application/json
   {
     "title": "Write tests",
     "description": "Add unit tests for Task API",
     "is_done": false
   }

Response:

http 
   HTTP/1.1 201 Created
   Content-Type: application/json
   {
     "id": 10,
     "title": "Write tests",
     "description": "Add unit tests for Task API",
     "is_done": false,
     "created_at": "2026-08-28T16:00:00Z",
     "updated_at": "2026-08-28T16:00:00Z"
   }
  1. Read the task
http 
   GET /tasks/10

Response:

http 
   HTTP/1.1 200 OK
   {
     "id": 10,
     "title": "Write tests",
     "description": "Add unit tests for Task API",
     "is_done": false,
     "created_at": "2026-08-28T16:00:00Z",
     "updated_at": "2026-08-28T16:00:00Z"
   }
  1. Update the task to mark it as done
http 
   PATCH /tasks/10
   Content-Type: application/json
   {
     "is_done": true
   }

Response:

http 
   HTTP/1.1 200 OK
   {
     "id": 10,
     "title": "Write tests",
     "description": "Add unit tests for Task API",
     "is_done": true,
     "created_at": "2026-08-28T16:00:00Z",
     "updated_at": "2026-08-28T17:15:00Z"
   }
  1. Delete the task
http 
   DELETE /tasks/10

Response:

http 
   HTTP/1.1 204 No Content
  1. Try to read it again, and get 404:
http 
   GET /tasks/10

Response:

http 
   HTTP/1.1 404 Not Found
   {
     "detail": "Task not found"
   }

Error Handling in CRUD Operations

Every CRUD endpoint should handle typical error cases:

OperationCommon error situationStatus code
CreateInvalid input data400 / 422
ReadTask not found404
UpdateTask not found404
UpdateInvalid field types or values400 / 422
DeleteTask not found404

FastAPI uses 422 Unprocessable Entity by default when the request body cannot be validated against the Pydantic model.

Example of raising your own error:

python 
if len(task_in.title.strip()) == 0:
    raise HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail="Title cannot be empty",
    )

Key error handling rule

  • For invalid input data, return a 4xx error and a clear message that helps the client fix the request.

Putting It All Together

A minimal, complete tasks router might look like this:

python 
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import List
from . import models, schemas
from .database import get_db
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.post("", response_model=schemas.TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(task_in: schemas.TaskCreate, db: Session = Depends(get_db)):
    task = models.Task(
        title=task_in.title,
        description=task_in.description,
        is_done=task_in.is_done,
    )
    db.add(task)
    db.commit()
    db.refresh(task)
    return task
@router.get("", response_model=List[schemas.TaskRead])
def list_tasks(
    db: Session = Depends(get_db),
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
):
    tasks = db.query(models.Task).offset(skip).limit(limit).all()
    return tasks
@router.get("/{task_id}", response_model=schemas.TaskRead)
def get_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task
@router.patch("/{task_id}", response_model=schemas.TaskRead)
def update_task(
    task_id: int,
    task_in: schemas.TaskUpdate,
    db: Session = Depends(get_db),
):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    update_data = task_in.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(task, key, value)
    db.commit()
    db.refresh(task)
    return task
@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    db.delete(task)
    db.commit()

Once you have these CRUD operations, you have a complete, usable Task Management API. In later chapters you will add validation layers, testing, authentication, and documentation on top of this foundation.

Views: 5

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3506 / 24h
Impressum / Imprint
Datenschutz / Data Policy