Implementing Backend Features - Part 2

!git checkout -b feature/crud-endpoints
/clear

Read @specs/api-contract.md.

Build the four FastAPI endpoints in backend/app/api/v1/tasks.py:
  GET    /api/tasks       - list all tasks
  POST   /api/tasks       - create a task (return 201)
  PUT    /api/tasks/{id}  - update a task
  DELETE /api/tasks/{id}  - delete a task (return 204)

Business logic goes in backend/app/services/task_service.py.
Routes call the service. Service talks to the database.
Routes know nothing about SQLAlchemy. Service knows nothing about HTTP.
Response shapes must match exactly what is in the spec.
Handle errors with HTTPException using the status codes from the spec.
Wire the router into backend/app/main.py with CORS enabled so the frontend can call it.

router = APIRouter(prefix="/api/tasks", tags=["tasks"])
DbDep = Annotated[AsyncSession, Depends(get_db)]


@router.get("", response_model=list[TaskResponse], response_model_by_alias=True)
async def list_tasks(db: DbDep) -> list[Task]:
    return await task_service.get_all_tasks(db)


@router.post("", response_model=TaskResponse, response_model_by_alias=True,
             status_code=status.HTTP_201_CREATED)
async def create_task(data: TaskCreate, db: DbDep) -> Task:
    return await task_service.create_task(db, data)


@router.put("/{task_id}", response_model=TaskResponse, response_model_by_alias=True)
async def update_task(task_id: int, data: TaskUpdate, db: DbDep) -> Task:
    task = await task_service.update_task(db, task_id, data)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int, db: DbDep) -> None:
    if not await task_service.delete_task(db, task_id):
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")

async def update_task(db: AsyncSession, task_id: int, data: TaskUpdate) -> Task | None:
    task = await db.get(Task, task_id)
    if task is None:
        return None

    # Only change the fields the client actually sent
    patch = data.model_dump(exclude_unset=True)
    for field, value in patch.items():
        setattr(task, field, value)

    await db.commit()
    await db.refresh(task)
    return task

Run the backend server with uv.

http://localhost:8000/docs

# List all tasks (200)
curl -s http://localhost:8000/api/tasks

# Create a task (201). Note the camelCase field names on the wire
curl -s -X POST http://localhost:8000/api/tasks \
  -H "Content-Type: application/json" \
  -d '{"title":"Docker basics","description":"Intro video","assignedRole":"Content","state":"code_ready"}'

# Move task 1 forward (200). Only the state is sent
curl -s -X PUT http://localhost:8000/api/tasks/1 \
  -H "Content-Type: application/json" \
  -d '{"state":"recorded"}'

# Invalid state is rejected by Pydantic (422)
curl -s -o /dev/null -w "%{http_code}\n" -X PUT http://localhost:8000/api/tasks/1 \
  -H "Content-Type: application/json" -d '{"state":"flying"}'

# Delete the task you created (204), then delete it again (404)
# Replace <id> with the id returned by the POST above
curl -s -o /dev/null -w "%{http_code}\n" -X DELETE http://localhost:8000/api/tasks/<id>
curl -s -o /dev/null -w "%{http_code}\n" -X DELETE http://localhost:8000/api/tasks/<id>

Merge this branch into main and then delete the feature/crud-endpoints branch.

!git checkout -b feature/seed-data

Read @specs/seed-data.json. Write a seed script at backend/seed.py
that reads this file and inserts all records into the tasks table
using the async SQLAlchemy session.

The script must be idempotent: if run twice, it should not create
duplicates. Check whether a task with the same id already exists
before inserting.

Run it with uv and confirm the database has five records.

async def seed() -> None:
    records = json.loads(SEED_FILE.read_text(encoding="utf-8"))
    async with AsyncSessionLocal() as db:
        for rec in records:
            if await db.get(Task, rec["id"]) is not None:
                print(f"  SKIP   id={rec['id']}")
                continue
            db.add(Task(
                id=rec["id"],
                title=rec["title"],
                description=rec["description"],
                assigned_role=AssignedRole(rec["assigned_role"]),
                state=TaskState(rec["state"]),
                created_at=date.fromisoformat(rec["created_at"]),
            ))
        await db.commit()

cd backend
uv run python seed.py

Merge this branch into main and delete the feature/seed-data branch.

/clear