Post Todo (User ID)

🔐 Protecting Todo Creation with JWT

Now that we can successfully authenticate a JWT using our get_current_user function, it's time to apply that authentication to our Todo endpoints.

In this lecture, we are going to:

• Connect authentication to our Todos router

• Protect the create_todo endpoint

• Attach the authenticated user to the todo (owner relationship)

• Test everything using Swagger

This is a critical backend concept: securing endpoints and linking data to authenticated users.

🧠 Step 1: Understanding get_current_user (Auth Layer)

Inside our auth.py, we created a function that:

• Reads the JWT token

• Decodes it

• Extracts the payload

• Returns user information

🗂️ routers/auth.py

async def get_current_user(token: Annotated[str, Depends(oauth2_bearer)]):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get('sub')
        user_id: int = payload.get('id')

        if username is None or user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail='Could not validate user'
            )

        return {'username': username, 'id': user_id}

    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='Could not validate user.'
        )

🔍 What This Does

• Validates the JWT

• Extracts sub (username) and id

• Ensures the token is legitimate

• Returns structured user data

This is server-side authentication logic — the client cannot fake this.

📂 Step 2: Import Authentication into todos.py

Now open your todos.py.

We need to import our authentication dependency:

🗂️ todos.py

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

🔗 Step 3: Create a User Dependency

Now we inject authentication into our route using dependency injection.

🗂️ todos.py

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

This line creates a reusable dependency that tells FastAPI to call get_current_user and inject its returned value (a dictionary) into routes.

Short breakdown

• dict → expected return type

• Depends(get_current_user) → run get_current_user to get the user

• Annotated → combines the type and dependency

🧠 Why This Is Powerful

This means:

• Before our endpoint runs…

• FastAPI will validate the JWT

• If authentication fails → the request never reaches the route

• If successful → we receive a user dictionary

This is how backend APIs enforce security.

🏗️ Step 4: Protect the Create Todo Endpoint

Now let's modify our POST route.

🗂️ todos.py

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

Notice:

user: user_dependency

This means our route will only execute if authentication succeeds.

🔎 Step 5: Validate the User

Even though dependency injection handles authentication, we still add an extra safety check.

🗂️ todos.py

@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())
    db.add(todo_model)
    db.commit()

🗄️ Step 6: Attach the Authenticated User to the Todo

Now comes the most important backend concept:

Our Todos model contains an owner_id foreign key that references the Users table.

We must assign it manually.

🗂️ todos.py

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

🧠 Why This Matters

• We are linking the todo to the authenticated user

• This enforces data ownership

• Each user can only own their own records

• This is foundational for multi-user backend systems

🚀 Step 7: Run the Server

In your terminal:

uvicorn main:app --reload

🌐 Step 8: Testing in Swagger UI

Open:

<http://127.0.0.1:8000/docs>

Swagger UI will show a 🔒 lock icon on the /todo endpoint.

Why?

Because we added authentication dependency.

❌ What Happens Without Authorization?

If you click Execute without logging in:

You will receive:

401 Unauthorized
Not authenticated

This is correct backend behavior.

🔐 Step 9: Authorize in Swagger

Click the 🔒 lock icon.

• Enter your login credentials

• Click Authorize

• Close the popup

Now try executing again.

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(db: db_dependency):
    return db.query(Todos).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()