Authentication Enhancements

🔐 Refining Authentication Routes & Improving API Structure

In the previous session, we implemented OAuth2PasswordBearer and created a get_current_user() dependency to validate JWT tokens.

In this session, we will:

Improve error handling in the /token endpoint
Organize authentication routes using APIRouter prefixes
Clean up redundant route paths
Update our OAuth2 configuration
Improve Swagger documentation structure

Let’s make our authentication module production-ready.

🚫 Improve Authentication Error Handling

Previously, when authentication failed, we returned a simple string:

🗂️ routers/auth.py

if not user:
    return "Failed Authentication"

This is not ideal for a secure API. Instead, we should raise a proper HTTP exception.

✅ Updated `/token` Endpoint

🗂️ routers/auth.py

from fastapi import HTTPException, status

@router.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: db_dependency
):

    user = authenticate_user(form_data.username, form_data.password, db)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate user"
        )

    token = create_access_token(
        user.username,
        user.id,
        timedelta(minutes=20)
    )

    return {
        "access_token": token,
        "token_type": "bearer"
    }

🔎 Why This Is Better

Returns a proper 401 Unauthorized
Aligns with REST standards
Prevents ambiguous responses
Improves API documentation clarity in Swagger

🗂 Organize Routes Using Prefix and Tags

Currently, all endpoints appear in a single group inside Swagger.

To improve clarity, we’ll group authentication endpoints separately.

🛠 Update APIRouter Configuration

Modify your router definition:

🗂️ routers/auth.py

router = APIRouter(
    prefix="/auth",
    tags=["auth"]
)

📌 What This Does

Adds /auth automatically to all routes in this router
Groups endpoints under the "auth" section in Swagger UI
Improves documentation organization

After refreshing Swagger, you’ll now see:

Much cleaner and more professional.

✂️ Remove Redundant Route Paths

Previously, your create user endpoint looked like this:

@router.post("/auth/")

Because we already defined prefix="/auth", this creates:

/auth/auth

That’s redundant.

✅ Corrected Create User Endpoint

Update it to:

#remove auth
@router.post("/auth")
@router.post("/")
async def create_user(
    db: db_dependency,
    create_user_request: CreateUserRequest):
    ...............
    ........

📌 Resulting Route

Now it correctly maps to:

POST /auth

🔄 Update OAuth2PasswordBearer Configuration

Previously, we configured:

oauth2_bearer = OAuth2PasswordBearer(tokenUrl="token")

But now that we added the /auth prefix, our token endpoint is located at:

/auth/token

So we must update the configuration.

✅ Updated Bearer Configuration

oauth2_bearer = OAuth2PasswordBearer(tokenUrl="auth/token")

🔍 Why This Is Important

OAuth2PasswordBearer tells FastAPI that the API will use Bearer token authentication.
tokenUrl="token" specifies the endpoint where users obtain the token (login endpoint).
FastAPI automatically adds an authentication mechanism to Swagger UI.

🗂️ routes/auth.py

from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from datetime import datetime, timedelta, timezone
from pydantic import BaseModel
from sqlalchemy.orm import Session
from starlette import status
from database import SessionLocal
from models import Users
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
from jose import jwt, JWTError

router = APIRouter(
    prefix="/auth",
    tags=["auth"]
)
SECRET_KEY = "735d62ded4a5f73eec7de13a6fa3dd309a575d8feaccb644cf34e4841a88317a"
ALGORITHM = "HS256"

# Password hashing setup
bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_bearer = OAuth2PasswordBearer(tokenUrl="auth/token")

# Pydantic schema for user creation
class CreateUserRequest(BaseModel):
    username: str
    email: str
    first_name: str
    last_name: str
    password: str
    role: str

class Token(BaseModel):
    access_token: str
    token_type: str

# Database dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db_dependency = Annotated[Session, Depends(get_db)]

def authenticate_user(username: str, password: str, db: Session):
    user = db.query(Users).filter(Users.username == username).first()

    if not user:
        return False

    if not bcrypt_context.verify(password, user.hashed_password):
        return False

    return user

def create_access_token(username: str, user_id: int, expires_delta: timedelta):
    encode = {
        "sub": username,
        "id": user_id
    }

    expires = datetime.now(timezone.utc) + expires_delta
    encode.update({"exp": expires})

    return jwt.encode(encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(
    token: Annotated[str, Depends(oauth2_bearer)]):
    try:
        # Decode the token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

        # Extract values from payload
        username: str = payload.get("sub")
        user_id: int = payload.get("id")

        # Validate payload data
        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"
        )

# Create user endpoint
@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_user(
    create_user_request: CreateUserRequest,
    db: db_dependency
		):

    # Create a new user model
    create_user_model = Users(
        email=create_user_request.email,
        username=create_user_request.username,
        first_name=create_user_request.first_name,
        last_name=create_user_request.last_name,
        role=create_user_request.role,
        hashed_password=bcrypt_context.hash(create_user_request.password),
        is_active=True
    )

    # Save to database
    db.add(create_user_model)
    db.commit()

@router.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: db_dependency):
    user = authenticate_user(form_data.username, form_data.password, db)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate user"
        )

    token = create_access_token(
        user.username,
        user.id,
        timedelta(minutes=20)
    )

    return {
        "access_token": token,
        "token_type": "bearer"
    }

🔐 Refining Authentication Routes & Improving API Structure

🚫 Improve Authentication Error Handling

✅ Updated /token Endpoint

🔎 Why This Is Better

🗂 Organize Routes Using Prefix and Tags

🛠 Update APIRouter Configuration

📌 What This Does

✂️ Remove Redundant Route Paths

✅ Corrected Create User Endpoint

📌 Resulting Route

🔄 Update OAuth2PasswordBearer Configuration

✅ Updated Bearer Configuration

🔍 Why This Is Important

✅ Updated `/token` Endpoint