Get All Todos (User ID) – FastAPI - Beginner to Pro

🔎 Retrieving Todos for the Authenticated User Only

Now that we can save a todo per authenticated user, the next step is just as important:

We must ensure that when a user retrieves todos, they only see their own records.

This is a fundamental backend principle called:

🔐 Data Isolation

Each authenticated user should only access data that belongs to them.

🧠 The Problem

Currently, our read_all endpoint looks like this:

🗂️ routers/todos.py

@router.get("/", status_code=status.HTTP_200_OK)
async def read_all(db: db_dependency):
    return db.query(Todos).all()

🚨 What’s Wrong Here?

This returns every todo in the database.

If 10 users exist → one user would see all 10 users’ todos.

That’s a major backend security flaw.

We must filter by owner_id.

🔗 Step 1: Inject the Authenticated User

First, we modify the route to include our user_dependency.

🗂️ routers/todos.py

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

Now this route:

Requires authentication
Injects the validated user
Prevents unauthorized access

Because remember:

user_dependency = Annotated[dict, Depends(get_current_user)]

If JWT validation fails → this endpoint never runs.

🗄️ Step 2: Filter Todos by Owner

Now instead of returning everything:

db.query(Todos).all()

We filter by owner_id.

✅ Correct Backend Filtering

🗂️ routers/todos.py

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

🧠 What’s Happening Here?

Let’s break it down like backend engineers:

user.get('id') → comes from decoded JWT
Todos.owner_id → foreign key in database
.filter(...) → SQL WHERE clause
.all() → return matching records only

🧾 Equivalent SQL Concept

This is essentially:

terminal

SELECT * FROM todos WHERE owner_id = current_user_id;

That is how real production systems isolate user data.

🚀 Step 3: Run the Server

If your application is not running:

terminal

uvicorn main:app --reload

🌐 Test in Swagger

Open:

browser

<http://127.0.0.1:8000/docs>

Go to the GET / endpoint.

❌ Without Authorization

If you click Try it out → Execute:

You will see:

401 Unauthorized
Not authenticated

This is correct behavior.

The endpoint requires a valid JWT.

🔐 Authorize

Click the 🔒 lock icon.

Enter your login credentials
Click Authorize
Close the popup

Now execute again.

✅ You should see todos which only belongs to you

🗂️ 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(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')

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

No change in other files