๐๏ธ 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:
The todo ID must match
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 --reloadOpen:
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()