Full CRUD Operations

You've created tasks with POST. Now you complete the picture with Read, Update, and Delete. Together—Create, Read, Update, Delete—these form CRUD, the foundation of data-driven APIs.

This matters for agents: every agent that manages state needs CRUD. When you build agent endpoints (Lesson 7), agents will create conversation sessions, read past messages, update conversation metadata, and delete expired sessions. The patterns you learn here transfer directly.

Why CRUD Is the Foundation

Consider what agents actually do:

• Memory agents create memories, read relevant ones, update importance scores, delete stale entries
• Task agents create tasks, read pending work, update status, delete completed items
• Session agents create conversations, read context, update metadata, delete expired sessions

Every one of these is CRUD. Master these four operations, and you can build the data layer for any agent.

HTTP Methods and CRUD

Each CRUD operation maps to an HTTP method:

Operation	HTTP Method	Endpoint Example	Description
Create	POST	POST /tasks	Create a new task
Read (all)	GET	GET /tasks	List all tasks
Read (one)	GET	GET /tasks/1	Get task with ID 1
Update	PUT	PUT /tasks/1	Update task with ID 1
Delete	DELETE	DELETE /tasks/1	Delete task with ID 1

Why these specific mappings? HTTP methods have semantics:

• GET is safe—it doesn't change server state. Browsers can cache GET responses.
• POST creates new resources. Not safe, not idempotent.
• PUT replaces a resource. Idempotent—calling it twice has the same effect as once.
• DELETE removes a resource. Also idempotent.

These semantics matter for agents. If an agent's HTTP call fails partway through, idempotent operations (PUT, DELETE) can be safely retried. Non-idempotent operations (POST) require more careful handling.

List All Tasks

The simplest read operation—return everything:

@app.get("/tasks")
def list_tasks():
    return tasks

But agents rarely want everything. Add filtering:

@app.get("/tasks")
def list_tasks(status: str | None = None):
    if status:
        return [t for t in tasks if t["status"] == status]
    return tasks

Now clients can:

• GET /tasks — All tasks
• GET /tasks?status=pending — Only pending tasks
• GET /tasks?status=completed — Only completed tasks

Output (GET /tasks?status=pending):

[
  {"id": 1, "title": "Learn FastAPI", "status": "pending"},
  {"id": 3, "title": "Deploy app", "status": "pending"}
]

For agents, this filtering is essential. A task agent doesn't want to process completed items—it filters for pending work.

Get Single Task

Retrieve one task by ID:

from fastapi import HTTPException

@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    for task in tasks:
        if task["id"] == task_id:
            return task
    raise HTTPException(status_code=404, detail="Task not found")

What happens when the task doesn't exist? We raise HTTPException with status 404. This is the standard HTTP response for "resource not found."

Test it:

• GET /tasks/1 → Returns task 1 (if it exists)
• GET /tasks/999 → Returns 404 error

Output (GET /tasks/999):

HTTP/1.1 404 Not Found
content-type: application/json

{"detail": "Task not found"}

Update Task: The Subtle Complexity

This is where students often get confused. Let's work through it carefully.

PUT replaces the entire resource. The client sends the complete new version:

class TaskUpdate(BaseModel):
    title: str
    description: str | None = None
    status: str | None = None

@app.put("/tasks/{task_id}")
def update_task(task_id: int, task_update: TaskUpdate):
    for task in tasks:
        if task["id"] == task_id:
            task["title"] = task_update.title
            if task_update.description is not None:
                task["description"] = task_update.description
            if task_update.status is not None:
                task["status"] = task_update.status
            return task
    raise HTTPException(status_code=404, detail="Task not found")

Why a new model TaskUpdate? It allows updating status, which TaskCreate doesn't include. The client sends:

{
  "title": "Updated title",
  "description": "New description",
  "status": "completed"
}

The confusing part: Why check if task_update.description is not None?

Without this check, sending {"title": "New"} would set description to None, erasing the existing description. We only update fields the client explicitly provides.

But wait—isn't that PATCH behavior? Yes, this is actually partial update logic. True PUT would require ALL fields. In practice, most APIs use PUT for what should be PATCH because PATCH has browser support issues historically. You'll see both patterns.

Delete Task

Remove a task from our list:

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    for i, task in enumerate(tasks):
        if task["id"] == task_id:
            tasks.pop(i)
            return {"message": "Task deleted", "id": task_id}
    raise HTTPException(status_code=404, detail="Task not found")

Some APIs return 204 No Content for deletes. We're returning a confirmation message, which helps with debugging.

The In-Memory Mutation Problem

Look at our update code:

task["title"] = task_update.title

We're mutating a dictionary inside a list. This works, but has problems:

Problem 1: No atomic updates. If two requests try to update the same task simultaneously, they might interleave:

Request A: reads task with status="pending"
Request B: reads task with status="pending"
Request A: sets status="in_progress"
Request B: sets status="completed" (expected: in_progress -> completed, actual: pending -> completed)

Problem 2: No rollback. If we update title, then description fails validation, the title change persists.

Problem 3: The delete index bug. Our delete uses tasks.pop(i). If two deletes run simultaneously on IDs 1 and 3, the indices shift and we might delete the wrong item.

These are real problems that databases solve. For learning CRUD, they don't matter. But when you build production agents, you'll use a database with proper transaction support. We cover this in Chapter 47.

Complete Implementation

Here's the full main.py with all CRUD operations:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Task API")

# Models
class TaskCreate(BaseModel):
    title: str
    description: str | None = None

class TaskUpdate(BaseModel):
    title: str
    description: str | None = None
    status: str | None = None

class TaskResponse(BaseModel):
    id: int
    title: str
    description: str | None
    status: str

# Storage
tasks: list[dict] = []
task_counter = 0

# CREATE
@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate):
    global task_counter
    task_counter += 1
    new_task = {
        "id": task_counter,
        "title": task.title,
        "description": task.description,
        "status": "pending"
    }
    tasks.append(new_task)
    return new_task

# READ (all)
@app.get("/tasks")
def list_tasks(status: str | None = None):
    if status:
        return [t for t in tasks if t["status"] == status]
    return tasks

# READ (one)
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    for task in tasks:
        if task["id"] == task_id:
            return task
    raise HTTPException(status_code=404, detail="Task not found")

# UPDATE
@app.put("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, task_update: TaskUpdate):
    for task in tasks:
        if task["id"] == task_id:
            task["title"] = task_update.title
            if task_update.description is not None:
                task["description"] = task_update.description
            if task_update.status is not None:
                task["status"] = task_update.status
            return task
    raise HTTPException(status_code=404, detail="Task not found")

# DELETE
@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    for i, task in enumerate(tasks):
        if task["id"] == task_id:
            tasks.pop(i)
            return {"message": "Task deleted", "id": task_id}
    raise HTTPException(status_code=404, detail="Task not found")

Hands-On Exercise

Test the complete CRUD cycle in Swagger UI:

Step 1: Create Tasks

POST /tasks
{"title": "Learn CRUD", "description": "Complete this lesson"}

POST /tasks
{"title": "Build API", "description": "Create a complete REST API"}

POST /tasks
{"title": "Test API", "description": "Verify all endpoints work"}

Step 2: List and Filter

GET /tasks           # See all 3 tasks
GET /tasks?status=pending  # All should be pending

Step 3: Read Single

GET /tasks/1         # First task
GET /tasks/999       # Should return 404

Step 4: Update

PUT /tasks/1
{"title": "Learn CRUD", "status": "completed"}

Step 5: Verify Update

GET /tasks/1         # Status should be "completed"
GET /tasks?status=completed  # Should show task 1
GET /tasks?status=pending    # Should show tasks 2 and 3

Step 6: Delete

DELETE /tasks/3      # Delete task 3
GET /tasks           # Should only show tasks 1 and 2

Challenge: Design a Status Workflow

Before looking at any solution, design validation rules yourself:

The Problem: Tasks should follow a workflow:

• New tasks start as pending
• pending → in_progress → completed (normal flow)
• Can go back from in_progress → pending (blocked)
• Cannot go from completed → anything (final state)
• Cannot skip states (pending → completed directly)

Think about:

• Where do you validate this? In the Pydantic model or the endpoint?
• What error message helps the user understand what went wrong?
• How do you store valid transitions in a maintainable way?

Implement it. Test with invalid transitions. Then compare with AI:

"I implemented task status workflow validation like this: [paste your code]. I put the logic in [model/endpoint] because [your reasoning]. Would a state machine pattern be cleaner?"

Common Mistakes

Mistake 1: Not returning the updated resource

# Wrong - returns nothing useful
@app.put("/tasks/{task_id}")
def update_task(task_id: int, task_update: TaskUpdate):
    for task in tasks:
        if task["id"] == task_id:
            task["title"] = task_update.title
            return {"message": "Updated"}  # Client doesn't know the new state

# Correct - return the updated resource
@app.put("/tasks/{task_id}")
def update_task(task_id: int, task_update: TaskUpdate):
    for task in tasks:
        if task["id"] == task_id:
            task["title"] = task_update.title
            return task  # Client sees the result

Why return the updated resource? The client might have stale data. Returning the current state avoids a follow-up GET.

Mistake 2: Using wrong HTTP method

# Wrong - POST shouldn't be used for updates
@app.post("/tasks/{task_id}/update")

# Correct - PUT for updates
@app.put("/tasks/{task_id}")

# Wrong - GET with side effects
@app.get("/tasks/{task_id}/delete")

# Correct - DELETE for deletion
@app.delete("/tasks/{task_id}")

Why this matters: HTTP semantics have meaning. Browsers might prefetch GET requests. Proxies might cache them. If your GET deletes data, you'll have mysterious data loss.

Mistake 3: Not handling not-found cases

# Wrong - returns None, causes errors
@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    for task in tasks:
        if task["id"] == task_id:
            return task
    # Falls through, returns None

# Correct - explicit 404
@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    for task in tasks:
        if task["id"] == task_id:
            return task
    raise HTTPException(status_code=404, detail="Task not found")

Try With AI

Now that you've implemented full CRUD, deepen your understanding with these explorations.

Prompt 1: Understand PUT vs PATCH

My FastAPI PUT endpoint acts like PATCH—it only updates fields that are provided:

@app.put("/tasks/{task_id}")
def update_task(task_id: int, task_update: TaskUpdate):
    task["title"] = task_update.title
    if task_update.status is not None:
        task["status"] = task_update.status

Is this technically wrong according to HTTP semantics? Show me what strict PUT would look like, then explain why most APIs choose partial updates anyway.

What you're learning: This prompt clarifies the PUT vs PATCH debate. You'll discover that strict PUT requires the full resource representation, but practical APIs use PUT for partial updates because PATCH had browser support issues historically. Understanding this helps you make informed API design decisions.

Prompt 2: Optimize Lookup Performance

My current task lookup is O(n):

for task in tasks:
    if task["id"] == task_id:
        return task

Show me how to restructure this using a dictionary for O(1) lookups. But I also need to list tasks in creation order—how do I maintain ordering while getting O(1) lookups? Compare the trade-offs.

What you're learning: This prompt develops your data structure intuition. You'll learn that dict gives O(1) access but loses ordering, while collections.OrderedDict or separate list+dict structures maintain both. This matters when agent systems need fast lookups AND chronological listings.

Prompt 3: Handle Race Conditions

I'm building an endpoint for agents to claim tasks:

@app.post("/tasks/{task_id}/claim")
def claim_task(task_id: int, worker_id: str):
    task = find_task(task_id)
    if task["status"] != "pending":
        raise HTTPException(400, "Task already claimed")
    task["status"] = "in_progress"
    task["worker_id"] = worker_id

Two agents call this simultaneously for the same task. What happens? Show me how to prevent both from claiming the same task.

What you're learning: This prompt introduces concurrency challenges. You'll discover that in-memory operations aren't atomic—two requests can both read "pending" before either writes. This is why production agent systems use databases with proper locking or optimistic concurrency control.

---

Reflect on Your Skill

You built a fastapi-agent skill in Lesson 0. Test and improve it based on what you learned.

Test Your Skill

Using my fastapi-agent skill, help me implement full CRUD operations for a resource.
Does my skill include proper HTTP methods (GET, POST, PUT, DELETE) and status codes?

Identify Gaps

Ask yourself:

• Did my skill include all CRUD operations with correct HTTP methods?
• Did it handle resource lookup patterns and 404 errors?
• Did it use appropriate status codes (200, 201, 204, 404)?

Improve Your Skill

If you found gaps:

My fastapi-agent skill is missing complete CRUD patterns.
Update it to include GET (list and single), POST, PUT, DELETE operations,
proper HTTP status codes, and resource not-found handling.