RESTful Design & JWT Authentication

In this phase we wire up the public face of TaskFlow: API endpoints that follow REST conventions, secured with JWT tokens and guarded by role-based access control.

RESTful URL Structure

A clean REST API maps resources to URLs and uses HTTP methods to express intent:

Method	URL Pattern	Purpose	Status Code
POST	/api/v1/projects	Create a project	201 Created
GET	/api/v1/projects	List projects	200 OK
GET	/api/v1/projects/{id}	Get one project	200 OK
PUT	/api/v1/projects/{id}	Update a project	200 OK
DELETE	/api/v1/projects/{id}	Delete a project	204 No Content
POST	/api/v1/projects/{id}/tasks	Create a task in project	201 Created

Rules of thumb:

• Use plural nouns for collections (/projects, not /project).
• Nest child resources under parents (/projects/{id}/tasks).
• Version your API (/api/v1/) so breaking changes never surprise clients.
• Return the correct status code: 201 for creation, 204 for deletion, 404 when not found, 403 when forbidden, 422 for validation errors.

JWT Authentication Flow

TaskFlow uses JSON Web Tokens (JWT) for stateless authentication:

Register        Login             Access Protected Route
────────►  ────────────►  ────────────────────────────────►
POST         POST              GET /api/v1/projects
/auth/       /auth/login       Authorization: Bearer <token>
register     ─► JWT token      ─► Server decodes token
                                  ─► Identifies user
                                  ─► Returns data

A JWT has three parts separated by dots: header.payload.signature. The server signs the token with a secret key; on every request, it verifies the signature without hitting the database.

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

SECRET_KEY = "your-secret-key"  # loaded from environment variable
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

Password Hashing with pwdlib + Argon2

FastAPI's current documentation recommends pwdlib with the Argon2 backend instead of the older passlib/bcrypt combination. Argon2 is the winner of the Password Hashing Competition and is resistant to GPU-based brute-force attacks:

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

password_hash = PasswordHash((Argon2Hasher(),))

hashed = password_hash.hash("user-password")           # hash
is_valid = password_hash.verify("user-password", hashed) # verify

FastAPI Dependency Injection for Auth

FastAPI's Depends() system lets you inject the current user into any endpoint. The dependency reads the Authorization: Bearer <token> header, decodes the JWT, looks up the user in the database, and returns the user object (or raises 401 Unauthorized):

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:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

    user = await db.get(User, int(user_id))
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return user

Any endpoint that needs authentication simply adds current_user: User = Depends(get_current_user) to its signature. FastAPI handles the rest.

Role-Based Access Control (RBAC)

In TaskFlow, every user has a role within a project (owner, admin, or member). Access checks happen through another dependency:

from enum import Enum

class ProjectRole(str, Enum):
    OWNER = "owner"
    ADMIN = "admin"
    MEMBER = "member"

def require_project_role(*allowed_roles: ProjectRole):
    async def checker(
        project_id: int,
        current_user: User = Depends(get_current_user),
        db: AsyncSession = Depends(get_db),
    ) -> ProjectMember:
        member = await get_project_member(db, project_id, current_user.id)
        if member is None or member.role not in allowed_roles:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
        return member
    return checker

Usage in an endpoint:

@router.put("/projects/{project_id}")
async def update_project(
    project_id: int,
    data: ProjectUpdate,
    member: ProjectMember = Depends(
        require_project_role(ProjectRole.OWNER, ProjectRole.ADMIN)
    ),
    db: AsyncSession = Depends(get_db),
):
    # only owner or admin reaches this line
    ...

Pagination Pattern

For list endpoints, TaskFlow uses page/size pagination with a total count so the frontend can render page controls:

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

Query parameters ?page=1&size=20 drive the offset calculation: offset = (page - 1) * size. The response always includes total (the full count) and pages (the total number of pages).

Automatic OpenAPI Docs

FastAPI generates interactive API documentation from your code at no extra effort:

• Swagger UI at /docs -- test endpoints directly in the browser.
• ReDoc at /redoc -- a clean, readable reference.

Pydantic v2 models, response status codes, and dependency-injected auth all appear automatically in the generated docs. Clients can export the OpenAPI JSON from /openapi.json to generate SDKs in any language.

---

Next: Hands-on lab -- you will build all auth, project, and task endpoints for TaskFlow. :::