Building Your First API — Task Manager (Part 3)¶

🎯 What You’ll Learn¶

How to extend your Task Manager with user accounts
How to enforce per-user task ownership using JWT
How to create admin-only endpoints with role checks
How to apply rate limiting to signup/login endpoints to prevent brute-force attacks

🧱 Step 1: User Model¶

📄 models/user.py

from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True, unique=True)
    email: str
    hashed_password: str
    is_active: bool = True
    role: str = "user"

    tasks: List["Task"] = Relationship(back_populates="user")

🧱 Step 2: Task Model with Ownership¶

📄 models/task.py

from sqlmodel import SQLModel, Field, Relationship
from typing import Optional
from models.user import User

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    description: Optional[str] = None
    completed: bool = False
    user_id: Optional[int] = Field(default=None, foreign_key="user.id")

    user: Optional[User] = Relationship(back_populates="tasks")

🔐 Step 3: Current User Dependency¶

📄 core/dependencies.py

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from core.jwt import decode_access_token
from sqlmodel import Session, select
from db import get_session
from models.user import User

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

def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)) -> User:
    payload = decode_access_token(token)
    if not payload or "sub" not in payload:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = session.exec(select(User).where(User.username == payload["sub"])).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

def require_admin(user: User = Depends(get_current_user)):
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return user

def require_active_user(user: User = Depends(get_current_user)):
    if not user.is_active:
        raise HTTPException(status_code=403, detail="Inactive account")
    return user

⚡ Step 4: Rate Limiting in `routers/auth.py`¶

Instead of configuring rate limiting globally in main.py, we apply it directly to signup and login routes.

📄 routers/auth.py

from fastapi import APIRouter, Depends, HTTPException, Request
from sqlmodel import Session, select
from models.user import User
from db import get_session
from core.security import hash_password, verify_password
from fastapi.security import OAuth2PasswordRequestForm
from core.jwt import create_access_token

# SlowAPI imports
from slowapi import Limiter
from slowapi.util import get_remote_address

router = APIRouter(prefix="/auth", tags=["Auth"])

# Create limiter instance for this router
limiter = Limiter(key_func=get_remote_address)

@router.post("/signup", status_code=201)
@limiter.limit("3/minute")   # limit signup attempts per IP
def signup(request: Request, user: User, session: Session = Depends(get_session)):
    existing_user = session.exec(select(User).where(User.username == user.username)).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="Username already exists")

    user.hashed_password = hash_password(user.hashed_password)
    # Add user role
    user.role = user.role or "user"  # default to "user"

    session.add(user)
    session.commit()
    session.refresh(user)
    return {"message": "User created", "user_id": user.id}

@router.post("/login")
@limiter.limit("5/minute")   # limit login attempts per IP
def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)):
    user = session.exec(select(User).where(User.username == form_data.username)).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    token = create_access_token({"sub": user.username})
    return {"access_token": token, "token_type": "bearer"}

Here we added Request because SlowAPI (the rate-limiting library you're using) needs to "inspect" the incoming connection to figure out who to block. For more details check the official documentation

📄 main.py:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from db import engine
from sqlmodel import SQLModel
from routers.auth import router as auth_router, limiter
from routers.tasks import router as tasks_router

from fastapi.middleware.cors import CORSMiddleware

from slowapi.errors import RateLimitExceeded
from slowapi import _rate_limit_exceeded_handler

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- Startup Logic ---
    # This runs before the application starts taking requests
    SQLModel.metadata.create_all(engine)

    yield  # The application runs while stuck here

    # --- Shutdown Logic ---
    # This runs after the application finishes handling requests
    # (e.g., close DB connections, clean up resources)
    pass

app = FastAPI(lifespan=lifespan)

# Add the exception handler to the MAIN app
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

app.include_router(auth_router)
app.include_router(tasks_router)

origins = [
    "http://localhost:3000",  # e.g. frontend dev server
    "https://myfrontend.com", # Production frontend
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,        # Which domains can access
    allow_credentials=True,       # Allow cookies/headers
    allow_methods=["*"],          # Which HTTP methods
    allow_headers=["*"],          # Which headers
)

🔍 Why this is better¶

Rate limiting is scoped to auth routes (not global).
Prevents brute-force attacks on login/signup.
Keeps main.py clean — only routers handle their own limits.

🛡️ Step 5: Secure Task Endpoints¶

Update the task service to handle User-Task relationships:

📄 services/task_service.py

from sqlmodel import Session, select
from models.task import Task
from models.user import User
from fastapi import HTTPException

def create_task(task: Task, session: Session, user: User) -> Task:
    task.user_id = user.id
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

def get_task_by_id(task_id: int, session: Session, user: User) -> Task:
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    elif task.user_id != user.id:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

def update_task(task_id: int, updated: Task, session: Session, user: User) -> Task:
    task = get_task_by_id(task_id, session, user)
    task.title = updated.title
    task.description = updated.description
    task.completed = updated.completed
    session.commit()
    session.refresh(task)
    return task

def delete_task(task_id: int, session: Session, user: User) -> None:
    task = get_task_by_id(task_id, session, user)
    session.delete(task)
    session.commit()

def list_tasks(session: Session, user: User, completed: bool | None = None) -> list[Task]:
    query = select(Task).where(Task.user_id == user.id)
    if completed is not None:
        query = query.where(Task.completed == completed)
    return session.exec(query).all()

def list_tasks_paginated(
    session: Session,
    user: User,
    completed: bool | None = None,
    limit: int = 10,
    offset: int = 0
) -> list[Task]:
    query = select(Task).where(Task.user_id == user.id)
    if completed is not None:
        query = query.where(Task.completed == completed)
    query = query.offset(offset).limit(limit)
    return session.exec(query).all()

Update the tasks router dependencies and secure task endpoints.

📄 routers/tasks.py

from fastapi import APIRouter, Depends
from sqlmodel import Session
from db import get_session
from models.task import Task
from models.user import User
from services import task_service
from core.dependencies import require_active_user, get_current_user

router = APIRouter(prefix="/tasks", tags=["Tasks"])

@router.get("/", dependencies=[Depends(require_active_user)])
def get_all(completed: bool | None = None, session: Session = Depends(get_session), user: User = Depends(get_current_user)):
    return task_service.list_tasks(session, user,  completed)

@router.get("/paginated", dependencies=[Depends(require_active_user)])
def get_paginated(
    completed: bool | None = None,
    limit: int = 10,
    offset: int = 0,
    session: Session = Depends(get_session),
    user: User = Depends(get_current_user)
):
    return task_service.list_tasks_paginated(session, user, completed, limit, offset)

@router.get("/{task_id}", dependencies=[Depends(require_active_user)])
def get_one(task_id: int, session: Session = Depends(get_session), user: User = Depends(get_current_user)):
    return task_service.get_task_by_id(task_id, session, user)

@router.post("/", status_code=201, dependencies=[Depends(require_active_user)])
def create(task: Task, session: Session = Depends(get_session), user: User = Depends(get_current_user)):
    return task_service.create_task(task, session, user)

@router.put("/{task_id}", dependencies=[Depends(require_active_user)])
def update(task_id: int, updated: Task, session: Session = Depends(get_session), user: User = Depends(get_current_user)):
    return task_service.update_task(task_id, updated, session, user)

@router.delete("/{task_id}", status_code=204, dependencies=[Depends(require_active_user)])
def delete(task_id: int, session: Session = Depends(get_session), user: User = Depends(get_current_user)):
    task_service.delete_task(task_id, session, user)

🛡️ Step 6: Admin-Only Endpoints¶

📄 routers/admin.py

from fastapi import APIRouter, Depends
from sqlmodel import Session, select
from db import get_session
from models.user import User
from core.dependencies import require_admin

router = APIRouter(prefix="/admin", tags=["Admin"])

@router.get("/users", dependencies=[Depends(require_admin)])
def list_users(session: Session = Depends(get_session)):
    return session.exec(select(User)).all()

🧪 Try It in Swagger UI¶

Try signing up more than 3 times in a minute → get rate limit error.
Try logging in more than 5 times in a minute → get rate limit error.
Regular users → only see their own tasks.
Admin users → access /admin/users to see all users.

🧠 Recap¶

You now have a secure Task Manager API with:

Feature	Implementation
User accounts	`User` model with roles
Task ownership	`user_id` foreign key in `Task`
Per-user access control	`get_current_user` dependency + filters
Admin-only endpoints	`require_admin` dependency
Rate limiting	Applied directly in `routers/auth.py`

🧪 Practice Challenge¶

Add a PATCH /users/{id}/role endpoint (admin-only) to promote/demote users.
Add a GET /auth/me endpoint that returns the current user’s profile.
Add rate limiting to /auth/me (e.g. 20 requests/minute) to prevent abuse.