🚀 Role-Based Authorization

So far:

All Todo endpoints require authentication.
Users can only access their own todos.
Our Users table includes a role column.

Now we’re going to implement:

✅ Admin can view all todos

✅ Admin can delete any user's todo

❌ Normal users cannot access these endpoints

This is called Role-Based Access Control (RBAC).

🗂 Step 1 — Our Users Model

Inside models.py:

🗂models.py

class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True)
    username = Column(String, unique=True)
    first_name = Column(String)
    last_name = Column(String)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    role = Column(String)

We made role a String because:

We may later have multiple roles
- "admin"
- "manager"
- "user"
- etc.

Currently, we manually assign "admin" inside the database.

🔐 Step 2 — Add Role to the JWT Token

Previously in auth.py, we encoded:

🗂auth.py

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

That’s not enough anymore.

We need the role inside the token.

Update `create_access_token`

🗂 routes/auth.py

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

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

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

Now the JWT contains:

username
user id
role

This is critical for authorization.

🔑 Step 3 — Pass Role When Creating Token

Inside the login endpoint: Update login_for_access_token 🗂 routes/auth.py

@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="Failed Authentication")

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

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

Now the token contains the user’s role.

🔍 Step 4 — Decode Role in `get_current_user`

Now we add role:

🗂 routes/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')
        user_role: str = payload.get('role')

        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,
            'role': user_role
        }

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

Now every endpoint that depends on get_current_user receives:

{
    "username": "...",
    "id": ...,
    "role": "admin" or "user"
}

This is powerful.

🛠 Step 5 — Create `admin.py` Router

Create a new file:

routers/admin.py

Copy Shared Dependencies

🗂 routes/admin.py

from typing import Annotated
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

Create Router

🗂 routes/admin.py

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

Database Dependency

🗂 routes/admin.py

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)]

📚 Step 6 — Admin: Read All Todos

routes/admin.py

@router.get("/todo", status_code=status.HTTP_200_OK)
async def read_all_todos(
    user: user_dependency,
    db: db_dependency
):
    if user is None or user.get("role") != "admin":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication Failed"
        )

    return db.query(Todos).all()

🔥 Important:

user.get("role") != "admin"

Only admins pass this check.

🗑 Step 7 — Admin: Delete Any Todo

🗂 routes/admin.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 or user.get("role") != "admin":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            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()

Now:

Admin can delete any todo
Normal users cannot

🧩 Step 8 — Register Admin Router in `main.py`

Inside main.py:

🗂 main.py

from routers import auth, todos, admin

Then include it:

🗂 main.py

app.include_router(auth.router)
app.include_router(todos.router)
app.include_router(admin.router)

🧪 Testing the Behaviour

👀 Before Testing

We created a new user with the following details:

{
  "username": "bhaskar",
  "email": "bhaskar@gmail.com",
  "first_name": "bhaskar",
  "last_name": "das",
  "password": "test1234",
  "role": ""
}

We also added 4 new todos to the database since there were none initially:

2 todos with username bhaskar
2 todos with username codeveda

Login as admin

/admin/todo → ✅ View all todos
- The default Get Todos endpoint / returns only the two todos belonging to the codeveda user.
- The Admin Get Todos endpoint returns all four todos.
/admin/todo/{id} → ✅ Allows deleting any todo
- In this example, we delete a todo that belongs to the bhaskar user.

Login as normal user

/admin/todo → ❌ 401 Unauthorized
Cannot delete other users' todos

🗂️ 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, role: str, expires_delta: timedelta):
    encode = {
        'sub': username,
        'id': user_id,
        'role': role
    }

    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:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

        username: str = payload.get('sub')
        user_id: int = payload.get('id')
        user_role: str = payload.get('role')

        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,
            'role': user_role
        }

    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="Failed Authentication")

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

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

🗂️ routes/admin.py

from typing import Annotated
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

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

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)]

@router.get("/todo", status_code=status.HTTP_200_OK)
async def read_all_todos(
    user: user_dependency,
    db: db_dependency
):
    if user is None or user.get("role") != "admin":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication Failed"
        )

    return db.query(Todos).all()

@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 or user.get("role") != "admin":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            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()

🗂️ main.py

from fastapi import FastAPI
import models
from database import engine
from routers import auth, todos, admin

app = FastAPI()

models.Base.metadata.create_all(bind=engine)

app.include_router(auth.router)
app.include_router(todos.router)
app.include_router(admin.router)

Admin Router

🚀 Role-Based Authorization

🗂 Step 1 — Our Users Model

🔐 Step 2 — Add Role to the JWT Token

Update create_access_token

🔑 Step 3 — Pass Role When Creating Token

🔍 Step 4 — Decode Role in get_current_user

🛠 Step 5 — Create admin.py Router

Copy Shared Dependencies

Create Router

Database Dependency

📚 Step 6 — Admin: Read All Todos

🗑 Step 7 — Admin: Delete Any Todo

🧩 Step 8 — Register Admin Router in main.py

🧪 Testing the Behaviour

👀 Before Testing

Login as admin

Login as normal user

Update `create_access_token`

🔍 Step 4 — Decode Role in `get_current_user`

🛠 Step 5 — Create `admin.py` Router

🧩 Step 8 — Register Admin Router in `main.py`