FastAPI - Beginner to Pro / Authenticate Requests

Delete Todo (User ID)

๐Ÿ—‘๏ธ Securing the DELETE Endpoint with Authentication

At this point, weโ€™ve secured:

  • โœ… Create

  • โœ… Read (All)

  • โœ… Read (Single)

  • โœ… Update

Now we will secure the final endpoint:

๐Ÿ” Delete a todo โ€” but only if it belongs to the authenticated user.

This completes full CRUD protection in our backend.


๐Ÿง  The Problem

Our original delete endpoint looks like this:

๐Ÿ—‚๏ธ routes/todos.py

@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.query(Todos).filter(Todos.id == todo_id).delete()
    db.commit()

๐Ÿšจ Whatโ€™s Wrong?

  • No authentication

  • No ownership validation

  • Any user could delete another userโ€™s todo

Thatโ€™s a critical backend vulnerability.


๐Ÿ” Step 1: Inject the Authenticated User

Just like our other endpoints, we inject user_dependency.

๐Ÿ—‚๏ธ routes/todos.py

@router.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(
    user: user_dependency,
    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.query(Todos).filter(Todos.id == todo_id).delete()
    db.commit()

Now:

  • JWT must be valid

  • Authentication runs before endpoint logic

  • We receive the authenticated user


๐Ÿ›ก๏ธ Step 2: Validate the User

We add our standard authentication check:

๐Ÿ—‚๏ธ routes/todos.py

@router.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_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)
		    .first()
		    
    if todo_model is None:
        raise HTTPException(status_code=404, detail='Todo not found.')

    db.query(Todos) \\
		    .filter(Todos.id == todo_id) \\
		    .delete()
		    
    db.commit()

This ensures:

  • If token is invalid โ†’ request fails

  • If token is missing โ†’ request fails

Consistent backend security pattern.


๐Ÿ—„๏ธ Step 3: Enforce Ownership When Fetching

Now we modify our query to include ownership validation.

๐Ÿ—‚๏ธ routes/todos.py

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

๐Ÿง  What This Does

We now enforce two conditions:

  1. The todo ID must match

  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;

๐Ÿ—‘๏ธ Step 4: Enforce Ownership During Delete

Even though we already validated ownership above, we also apply the same filters in the delete query.

This ensures database-level consistency.

๐Ÿ—‚๏ธ routes/todos.py

db.query(Todos) \\
    .filter(Todos.id == todo_id) \\
    .filter(Todos.owner_id == user.get('id')) \\
    .delete()

db.commit()

๐Ÿ“Œ Final Secure Version

๐Ÿ—‚๏ธ routes/todos.py

@router.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_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 None:
        raise HTTPException(status_code=404, detail='Todo not found.')

    db.query(Todos) \\
        .filter(Todos.id == todo_id) \\
        .filter(Todos.owner_id == user.get('id')) \\
        .delete()

    db.commit()

๐Ÿงช Testing the DELETE Endpoint

Make sure your server is running:

terminal

uvicorn main:app --reload

Open:

browser

<http://127.0.0.1:8000/docs>

Authenticate and Test


๐Ÿ—‚๏ธ routes/todos.py

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(
    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 None:
        raise HTTPException(status_code=404, detail='Todo not found.')

    db.query(Todos) \\
        .filter(Todos.id == todo_id) \\
        .filter(Todos.owner_id == user.get('id')) \\
        .delete()

    db.commit()