Put Todo (User ID)

✏️ Securing the Update Endpoint (PUT) with Authentication

Now that we can:

• ✅ Create todos per user

• ✅ Retrieve all todos per user

• ✅ Retrieve a single todo per user

The next step is just as important:

🔐 Secure the update (PUT) endpoint so users can only update their own todos.

This is how real backend systems prevent unauthorized modifications.

🧠 The Problem

Our original update endpoint looks like this:

🗂️ routers/todos.py

@app.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(
    db: db_dependency,
    todo_request: TodoRequest,
    todo_id: int = Path(gt=0)):
    todo_model = db.query(Todos).filter(Todos.id == todo_id).first()

    if todo_model is None:
        raise HTTPException(status_code=404, detail='Todo not found.')

    todo_model.title = todo_request.title
    todo_model.description = todo_request.description
    todo_model.priority = todo_request.priority
    todo_model.complete = todo_request.complete

    db.add(todo_model)
    db.commit()

🚨 What’s Wrong?

• No authentication

• No ownership validation

• Any user could update any todo if they know the ID

That is a serious backend security vulnerability.

🔐 Step 1: Inject the Authenticated User

We start by adding user_dependency.

🗂️ routers/todos.py

@app.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(
    user: user_dependency,
    db: db_dependency,
    todo_request: TodoRequest,
    todo_id: int = Path(gt=0)):
    
    todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
    if todo_model is None:
        raise HTTPException(status_code=404, detail='Todo not found.')

    todo_model.title = todo_request.title
    todo_model.description = todo_request.description
    todo_model.priority = todo_request.priority
    todo_model.complete = todo_request.complete

    db.add(todo_model)
    db.commit()

Now this endpoint:

• Requires a valid JWT

• Automatically validates authentication

• Injects the authenticated user

If authentication fails → request stops immediately.

🛡️ Step 2: Validate the User

Even though dependency injection handles authentication, we add an explicit check:

🗂️ routers/todos.py

@app.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(
    user: user_dependency,
    db: db_dependency,
    todo_request: TodoRequest,
    todo_id: int = Path(gt=0)):
    
    if user is None:
	    raise HTTPException(status_code=401, detail='Authentication Failed')
    
    todo_model = (
		db.query(Todos)
		.filter(Todos.id == todo_id)
		.first())
    if todo_model is None:
        raise HTTPException(status_code=404, detail='Todo not found.')

    todo_model.title = todo_request.title
    todo_model.description = todo_request.description
    todo_model.priority = todo_request.priority
    todo_model.complete = todo_request.complete

    db.add(todo_model)
    db.commit()

This enforces backend consistency across all protected endpoints.

🗄️ Step 3: Enforce Ownership in the Database Query

Previously we only filtered by ID:

🗂️ routers/todos.py

todo_model = (
		db.query(Todos)
		.filter(Todos.id == todo_id)
		.first())

Now we add ownership validation:

🗂️ routers/todos.py

todo_model = (
    db.query(Todos)
    .filter(Todos.id == todo_id)
    .filter(Todos.owner_id == user.get('id'))
    .first())

🧠 What This Does

We now enforce two conditions:

1. The ID must match the path parameter

2. The owner_id must match the authenticated user

Equivalent SQL Concept

SELECT * FROM todos
WHERE id = :todo_id
AND owner_id = :current_user_id;

If either condition fails → no record is returned.

That prevents:

• Cross-user updates

• ID guessing attacks

• Unauthorized data manipulation

📌 Final Secure Version

🗂️ routers/todos.py

@router.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(
    user: user_dependency,
    db: db_dependency,
    todo_request: TodoRequest,
    todo_id: int = Path(gt=0)):
    
    if user is None:
        raise HTTPException(status_code=401, detail='Authentication Failed')
    
    todo_model = (
    db.query(Todos)
    .filter(Todos.id == todo_id)
    .filter(Todos.owner_id == user.get('id'))
    .first())
    
    if todo_model is None:
        raise HTTPException(status_code=404, detail='Todo not found.')

    todo_model.title = todo_request.title
    todo_model.description = todo_request.description
    todo_model.priority = todo_request.priority
    todo_model.complete = todo_request.complete

    db.add(todo_model)
    db.commit()

🧪 Testing the Endpoint

Make sure your server is running:

terminal

uvicorn main:app --reload

Open:

browser

<http://127.0.0.1:8000/docs>

Authenticate and Test

from typing import Annotated
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from fastapi import APIRouter, Path, Depends, HTTPException
from starlette import status
from models import Todos
from database import SessionLocal
from .auth import get_current_user

router = APIRouter()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db_dependency = Annotated[Session, Depends(get_db)]
user_dependency = Annotated[dict, Depends(get_current_user)]

class TodoRequest(BaseModel):
    title: str = Field(min_length=3)
    description: str = Field(min_length=3, max_length=100)
    priority: int = Field(gt=0, lt=6)
    complete: bool

@router.get("/", status_code=status.HTTP_200_OK)
async def read_all(user: user_dependency, db: db_dependency):
    return db.query(Todos).filter(
        Todos.owner_id == user.get('id')).all()

@router.get("/todo/{todo_id}", status_code=status.HTTP_200_OK)
async def read_todo(user: user_dependency, db: db_dependency, todo_id: int = Path(gt=0)):

    if user is None:
        raise HTTPException(status_code=401, detail='Authentication Failed')

    todo_model = db.query(Todos).filter(Todos.id == todo_id).filter(Todos.owner_id == user.get('id')).first()
    if todo_model is not None:
        return todo_model
    else:
        raise HTTPException(status_code=404, detail='Todo Not Found')

@router.post("/todo/", status_code=status.HTTP_201_CREATED)
async def create_todo( user: user_dependency, db: db_dependency, todo_request: TodoRequest):

    if user is None:
        raise HTTPException(status_code=401, detail='Authentication Failed')

    todo_model = Todos(**todo_request.model_dump(), owner_id=user.get('id'))
    db.add(todo_model)
    db.commit()
    
@router.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(
    user: user_dependency,
    db: db_dependency,
    todo_request: TodoRequest,
    todo_id: int = Path(gt=0)):

    if user is None:
        raise HTTPException(status_code=401, detail='Authentication Failed')

    todo_model = db.query(Todos).filter(Todos.id == todo_id).filter(Todos.owner_id == user.get('id')).first()

    if todo_model is None:
        raise HTTPException(status_code=404, detail="Todo not found.")

    todo_model.title = todo_request.title
    todo_model.description = todo_request.description
    todo_model.priority = todo_request.priority
    todo_model.complete = todo_request.complete

    db.commit()
    
@router.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(
    db: db_dependency,
    todo_id: int = Path(gt=0)):
    todo_model = db.query(Todos).filter(Todos.id == todo_id).first()

    if todo_model is None:
        raise HTTPException(status_code=404, detail="Todo not found.")

    db.delete(todo_model)
    db.commit()