Get Todo (ID + User ID)

🔎 Securing “Get Todo by ID” with User Validation

Previously, we filtered all todos by the authenticated user.

Now we are going one level deeper:

🔐 Retrieve a single todo by ID — but only if it belongs to the authenticated user.

This is a critical backend security pattern:

Even if someone guesses another todo’s ID, they must not be able to access it.

🧠 The Problem

Our current read_todo endpoint looks like this:

🗂️ routers/todos.py

@router.get("/todo/{todo_id}", status_code=status.HTTP_200_OK)
async def read_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 not None:
        return todo_model
    else:
        raise HTTPException(status_code=404, detail='Todo Not Found')

🚨 What’s Missing?

• No authentication

• No user validation

• No ownership filtering

This means:

If User A knows User B’s todo ID → they could access it.

That’s a serious backend flaw.

🔐 Step 1: Inject the Authenticated User

Just like we did before, we must inject user_dependency.

🗂️ routers/todos.py

@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)):
		todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
		
		if todo_model is not None:
			return todo_model
		else:
			raise HTTPException(status_code=404, detail='Todo Not Found')

Now:

• JWT is validated first

• If authentication fails → route never executes

• If successful → we receive the user dictionary

🛡️ Step 2: Validate the User

Even though dependency injection already protects the route, we still add an explicit safety check.

🗂️ routers/todos.py

@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).first()
    if todo_model is not None:
        return todo_model
    else:
        raise HTTPException(status_code=404, detail='Todo Not Found')

This reinforces our backend security pattern.

🗄️ Step 3: Add Ownership Filtering

Previously, we only filtered by todo_id.

Now we add a second filter:

🗂️ routers/todos.py

@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')

🧠 What’s Happening Here?

We are now querying the database with two conditions:

1. The todo 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;

This ensures:

• Users can only fetch their own todos

• Cross-user access is blocked

• Even valid IDs won’t return results if ownership doesn’t match

This is real-world backend protection.

📌 Final Secure Version

🗂️ routers/todos.py

@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')

🔄 Applying the Same Security to read_all

We also reinforce validation in our read_all endpoint:

🗂️ routers/todos.py

@router.get("/", status_code=status.HTTP_200_OK)
async def read_all(user: user_dependency, db: db_dependency):

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

    return db.query(Todos).filter(
        Todos.owner_id == user.get('id')).all()

Now both endpoints are protected consistently.

Consistency is extremely important in backend architecture.

🚀 Step 4: Test the Application

Make sure your server is running:

uvicorn main:app --reload

Open:

<http://127.0.0.1:8000/docs>

🔐 Authenticate First

Click the 🔒 lock icon.

Authorize with your Bearer token.

If you’re already logged in, Swagger keeps your session active across endpoints.

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(
    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.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()