Build API Endpoints & Authentication

Objective

Build the full REST API layer for TaskFlow: authentication endpoints, project CRUD, and task management. Every endpoint must use async SQLAlchemy, Pydantic v2 request/response models, proper HTTP status codes, and JWT-based authorization.

Prerequisites

You should have the database models and async session from Phase 2. Your project structure should look like this:

taskflow/
├── app/
│   ├── core/
│   │   ├── config.py          # Settings (SECRET_KEY, DATABASE_URL, etc.)
│   │   ├── database.py        # async engine + get_db dependency
│   │   └── security.py        # ← NEW: JWT + password hashing
│   ├── models/
│   │   ├── user.py
│   │   ├── project.py
│   │   └── task.py
│   ├── schemas/               # ← NEW: Pydantic v2 models
│   │   ├── auth.py
│   │   ├── project.py
│   │   ├── task.py
│   │   └── common.py
│   ├── api/                   # ← NEW: route handlers
│   │   ├── deps.py            # get_current_user, require_project_role
│   │   ├── auth.py            # /api/v1/auth/*
│   │   ├── projects.py        # /api/v1/projects/*
│   │   └── tasks.py           # /api/v1/projects/{id}/tasks/* & /api/v1/tasks/*
│   └── main.py                # FastAPI app with router includes
├── requirements.txt
└── .env

Part 1 — Security Utilities (app/core/security.py)

Create the password hashing and JWT token utilities.

Password hashing — use pwdlib with Argon2:

from pwdlib import PasswordHash
from pwdlib.hashers.argon2 import Argon2Hasher

password_hash = PasswordHash((Argon2Hasher(),))

def hash_password(password: str) -> str:
    return password_hash.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)

JWT tokens — use python-jose[cryptography]:

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError

def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str:
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=30))
    to_encode = {"sub": subject, "exp": expire}
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

Part 2 — Pydantic v2 Schemas (app/schemas/)

Create request and response models. All models must use Pydantic v2 syntax (model_config = ConfigDict(from_attributes=True)).

app/schemas/common.py — shared pagination response:

from pydantic import BaseModel
from typing import Generic, TypeVar, Sequence

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: Sequence[T]
    total: int
    page: int
    size: int
    pages: int

app/schemas/auth.py — auth request/response models:

• UserCreate: email (EmailStr), password (min 8 chars), full_name
• UserResponse: id, email, full_name, created_at — with model_config = ConfigDict(from_attributes=True)
• Token: access_token, token_type

app/schemas/project.py — project models:

• ProjectCreate: name (1-100 chars), description (optional)
• ProjectUpdate: name (optional), description (optional)
• ProjectResponse: id, name, description, owner_id, created_at, updated_at

app/schemas/task.py — task models:

• TaskCreate: title (1-200 chars), description (optional), priority (low/medium/high/urgent), assigned_to (optional user id)
• TaskUpdate: title (optional), description (optional), status (todo/in_progress/done), priority (optional), assigned_to (optional)
• TaskResponse: id, title, description, status, priority, assigned_to, project_id, created_by, created_at, updated_at

Part 3 — Auth Dependencies (app/api/deps.py)

Create the FastAPI dependencies that protect your endpoints.

get_current_user — decodes the JWT Bearer token and returns the User from the database. Raises 401 Unauthorized if the token is missing, expired, or the user does not exist.

require_project_role(*allowed_roles) — returns a dependency that checks the current user's role in a given project. Raises 403 Forbidden if the user is not a member or their role is not in the allowed list.

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await db.get(User, int(user_id))
    if user is None:
        raise credentials_exception
    return user

Part 4 — Auth Router (app/api/auth.py)

Implement three endpoints:

POST /api/v1/auth/register

• Request body: UserCreate
• Check if email already exists → 409 Conflict
• Hash password, create user in database
• Return UserResponse with status 201

POST /api/v1/auth/login

• Accept OAuth2PasswordRequestForm (username = email, password)
• Verify email exists and password matches → 401 Unauthorized if not
• Return Token with the JWT access token

GET /api/v1/auth/me

• Requires authentication (get_current_user dependency)
• Return UserResponse for the current user

Part 5 — Projects Router (app/api/projects.py)

Implement four endpoints:

POST /api/v1/projects

• Requires authentication
• Request body: ProjectCreate
• Create project with owner_id = current_user.id
• Create a ProjectMember record with role owner
• Return ProjectResponse with status 201

GET /api/v1/projects

• Requires authentication
• Query params: page (default 1), size (default 20, max 100)
• Return only projects where the current user is a member
• Return PaginatedResponse[ProjectResponse]

GET /api/v1/projects/{project_id}

• Requires authentication + project membership (any role)
• Return ProjectResponse
• Return 404 if not found, 403 if not a member

PUT /api/v1/projects/{project_id}

• Requires authentication + project role: owner or admin
• Request body: ProjectUpdate
• Return updated ProjectResponse

DELETE /api/v1/projects/{project_id}

• Requires authentication + project role: owner only
• Delete project and all associated tasks/members
• Return 204 No Content

Part 6 — Tasks Router (app/api/tasks.py)

Implement three endpoints:

POST /api/v1/projects/{project_id}/tasks

• Requires authentication + project membership (any role)
• Request body: TaskCreate
• Set created_by = current_user.id, project_id = project_id, status = "todo"
• Return TaskResponse with status 201

GET /api/v1/projects/{project_id}/tasks

• Requires authentication + project membership (any role)
• Query params: page, size, status (filter), priority (filter), assigned_to (filter)
• Build dynamic SQLAlchemy query with optional filters
• Return PaginatedResponse[TaskResponse]

Example request:

GET /api/v1/projects/1/tasks?page=1&size=20&status=todo&priority=high

PUT /api/v1/tasks/{task_id}

• Requires authentication + membership in the task's project (any role)
• Look up task by id → 404 if not found
• Verify user is a member of the task's project → 403 if not
• Request body: TaskUpdate
• Return updated TaskResponse

DELETE /api/v1/tasks/{task_id}

• Requires authentication + project role: admin or owner on the task's project
• Return 204 No Content

Part 7 — Wire Up the App (app/main.py)

Register all three routers under the /api/v1 prefix:

from fastapi import FastAPI
from app.api import auth, projects, tasks

app = FastAPI(title="TaskFlow API", version="1.0.0")

app.include_router(auth.router, prefix="/api/v1/auth", tags=["auth"])
app.include_router(projects.router, prefix="/api/v1/projects", tags=["projects"])
app.include_router(tasks.router, prefix="/api/v1", tags=["tasks"])

Part 8 — Test with Swagger UI

Start the server and open http://localhost:8000/docs. Walk through this sequence:

1. Register a user via POST /api/v1/auth/register
2. Login via POST /api/v1/auth/login — copy the access_token
3. Click Authorize in Swagger UI, paste the token
4. Create a project via POST /api/v1/projects
5. List projects via GET /api/v1/projects
6. Create a task via POST /api/v1/projects/{id}/tasks
7. List tasks with filters via GET /api/v1/projects/{id}/tasks?status=todo

What to Submit

Your submission should contain 10 file sections in the editor below. Each section begins with a # FILE N: header.

---

Hints

• Use select().where(...).offset(...).limit(...) for paginated queries
• Use func.count() to get the total count efficiently
• For filtering, chain .where() calls conditionally:

  query = select(Task).where(Task.project_id == project_id)
  if status:
      query = query.where(Task.status == status)
  if priority:
      query = query.where(Task.priority == priority)
  

• Use math.ceil(total / size) to calculate the number of pages
• Remember to commit the session after create/update/delete operations