Chapter 18: Authorization — Protecting Routes and Verifying Current User -

Let’s Start With a Real Problem

Your blogging API is live. Users can log in and create posts. One day, a user named Priya notices something alarming: she can send a DELETE /posts/47 request — a post that belongs to a completely different user, someone she’s never even met — and it just… works. Your get_current_user dependency from Chapters 15 and 16 correctly confirmed who Priya is. But nothing ever checked whether Priya was actually allowed to delete post 47.

This is the difference between authentication (“who are you?”) and authorization (“what are you allowed to do?”). Chapter 16 gave us role-based authorization (regular user vs. admin). This chapter covers something just as common and just as important: ownership-based authorization — making sure a user can only modify their own data, not anyone else’s.

What You’ll Learn

The precise difference between authentication and authorization
How to check that the current user actually owns a resource before letting them modify or delete it
How to combine ownership checks with admin overrides (an admin can act on anyone’s data; a regular user only their own)
How to turn a repeated ownership check into a single, reusable FastAPI dependency
Theory + hands-on assignment questions to lock in what you’ve learned

Theory: Authentication vs. Authorization

	Authentication	Authorization
Question it answers	“Who is making this request?”	“Is this specific person allowed to do this specific thing?”
Where we built it	Chapters 14–16 (`get_current_user`)	Chapter 16 (roles) and this chapter (ownership)
Failure status code	`401 Unauthorized`	`403 Forbidden`

A request can be perfectly authenticated (a real, valid, logged-in user) and still be completely unauthorized for the specific action it’s attempting. Both checks matter, and they’re genuinely separate steps — get_current_user should only ever answer “who is this?”; a different piece of logic should answer “are they allowed to touch this particular row of data?”

Program 1: Basic Ownership Checks

Program Objective

Build a small, standalone FastAPI project where users can create posts, and only the post’s original author (or an admin) can update or delete it — with the ownership check written directly and explicitly inside each route.

Step-by-Step Implementation

Step 1: Create the project

uv init ownership-checks-demo
cd ownership-checks-demo
uv add fastapi --extra standard
uv add pydantic-settings
uv add sqlalchemy
uv add "passlib[bcrypt]"
uv add "python-jose[cryptography]"

Step 2: Create the project structure

mkdir -p app/core app/db app/models app/schemas app/services app/routers
touch app/__init__.py app/core/__init__.py app/db/__init__.py app/models/__init__.py app/schemas/__init__.py app/services/__init__.py app/routers/__init__.py

ownership-checks-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   ├── security.py
│   │   └── dependencies.py
│   ├── db/
│   │   ├── __init__.py
│   │   └── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── post.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── token.py
│   │   └── post.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── post_service.py
│   └── routers/
│       ├── __init__.py
│       ├── users.py
│       ├── auth.py
│       └── posts.py
├── pyproject.toml
└── uv.lock

Step 3: Create each file with the code below. Step 4: Run and test.

Complete Working Source Code

app/core/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Ownership Checks Demo"
    app_version: str = "1.0.0"
    database_url: str = "sqlite:///./app.db"
    jwt_secret_key: str = "a-very-insecure-demo-secret-change-this"
    jwt_algorithm: str = "HS256"
    access_token_expire_minutes: int = 30


settings = Settings()

app/core/security.py

from datetime import datetime, timedelta, timezone

from jose import JWTError, jwt
from passlib.context import CryptContext

from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(plain_password: str) -> str:
    return pwd_context.hash(plain_password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)


def create_access_token(subject: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
    to_encode = {"sub": subject, "exp": expire}
    return jwt.encode(to_encode, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)


def decode_token(token: str) -> dict:
    try:
        return jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm])
    except JWTError as exc:
        raise ValueError("Invalid or expired token.") from exc

app/core/dependencies.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from app.core.security import decode_token
from app.db.database import get_db
from app.models.user import User
from app.services.user_service import get_user_by_username

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")


def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials.",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        payload = decode_token(token)
    except ValueError:
        raise credentials_exception

    username = payload.get("sub")
    if username is None:
        raise credentials_exception

    user = get_user_by_username(db, username)
    if user is None:
        raise credentials_exception

    return user

app/db/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app.core.config import settings

engine = create_engine(settings.database_url, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


class Base(DeclarativeBase):
    pass


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

app/models/user.py

from sqlalchemy import Boolean, String
from sqlalchemy.orm import Mapped, mapped_column

from app.db.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(30), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_admin: Mapped[bool] = mapped_column(Boolean, default=False)

app/models/post.py

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column

from app.db.database import Base


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String(2000))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

app/schemas/user.py

from typing import Annotated

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserCreate(BaseModel):
    username: Annotated[str, Field(min_length=3, max_length=30)]
    email: Annotated[EmailStr, Field()]
    password: Annotated[str, Field(min_length=8, max_length=64)]


class UserOut(BaseModel):
    id: int
    username: str
    email: EmailStr
    is_admin: bool

    model_config = ConfigDict(from_attributes=True)

app/schemas/token.py

from pydantic import BaseModel


class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"

app/schemas/post.py

from typing import Annotated

from pydantic import BaseModel, ConfigDict, Field


class PostCreate(BaseModel):
    title: Annotated[str, Field(min_length=3, max_length=200)]
    content: Annotated[str, Field(min_length=1, max_length=2000)]


class PostPatch(BaseModel):
    title: Annotated[str | None, Field(default=None, min_length=3, max_length=200)]
    content: Annotated[str | None, Field(default=None, min_length=1, max_length=2000)]


class PostOut(BaseModel):
    id: int
    title: str
    content: str
    user_id: int

    model_config = ConfigDict(from_attributes=True)

app/services/user_service.py

from sqlalchemy.orm import Session

from app.core.security import hash_password
from app.models.user import User
from app.schemas.user import UserCreate


def create_user(db: Session, user_in: UserCreate) -> User:
    new_user = User(
        username=user_in.username,
        email=user_in.email,
        hashed_password=hash_password(user_in.password),
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


def get_user_by_username(db: Session, username: str) -> User | None:
    return db.query(User).filter(User.username == username).first()

app/services/post_service.py

from sqlalchemy.orm import Session

from app.models.post import Post
from app.schemas.post import PostCreate


def create_post(db: Session, post_in: PostCreate, owner_id: int) -> Post:
    new_post = Post(title=post_in.title, content=post_in.content, user_id=owner_id)
    db.add(new_post)
    db.commit()
    db.refresh(new_post)
    return new_post


def get_post_by_id(db: Session, post_id: int) -> Post | None:
    return db.get(Post, post_id)

app/routers/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from app.core.security import create_access_token, verify_password
from app.db.database import get_db
from app.schemas.token import TokenResponse
from app.services.user_service import get_user_by_username

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


@router.post("/login", response_model=TokenResponse)
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = get_user_by_username(db, form_data.username)

    if user is None or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password.")

    return TokenResponse(access_token=create_access_token(subject=user.username))

app/routers/users.py

from fastapi import APIRouter, Depends, status
from sqlalchemy.orm import Session

from app.db.database import get_db
from app.schemas.user import UserCreate, UserOut
from app.services.user_service import create_user

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


@router.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def register_user(user_in: UserCreate, db: Session = Depends(get_db)):
    return create_user(db, user_in)

app/routers/posts.py

from fastapi import APIRouter, Depends, HTTPException, Response, status
from sqlalchemy.orm import Session

from app.core.dependencies import get_current_user
from app.db.database import get_db
from app.models.user import User
from app.schemas.post import PostCreate, PostOut, PostPatch
from app.services.post_service import create_post, get_post_by_id

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


@router.post("/", response_model=PostOut, status_code=status.HTTP_201_CREATED)
def create_new_post(
    post_in: PostCreate,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    return create_post(db, post_in, owner_id=current_user.id)


@router.get("/{post_id}", response_model=PostOut)
def get_post(post_id: int, db: Session = Depends(get_db)):
    post = get_post_by_id(db, post_id)
    if post is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found.")
    return post


@router.patch("/{post_id}", response_model=PostOut)
def update_post(
    post_id: int,
    data: PostPatch,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    post = get_post_by_id(db, post_id)
    if post is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found.")

    # --- Authorization check: must be the owner, or an admin ---
    if post.user_id != current_user.id and not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not have permission to modify this post.",
        )

    update_data = data.model_dump(exclude_unset=True)
    for field_name, value in update_data.items():
        setattr(post, field_name, value)

    db.commit()
    db.refresh(post)
    return post


@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(
    post_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    post = get_post_by_id(db, post_id)
    if post is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found.")

    # --- Authorization check: must be the owner, or an admin ---
    if post.user_id != current_user.id and not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not have permission to delete this post.",
        )

    db.delete(post)
    db.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

app/main.py

from fastapi import FastAPI

from app.core.config import settings
from app.db.database import Base, engine
from app.models import post, user  # noqa: F401
from app.routers import auth, posts, users

Base.metadata.create_all(bind=engine)

app = FastAPI(title=settings.app_name, version=settings.app_version)

app.include_router(users.router)
app.include_router(auth.router)
app.include_router(posts.router)


@app.get("/", tags=["Root"])
def read_root():
    return {"message": f"Welcome to {settings.app_name}"}

Detailed Code Explanation

GET /posts/{post_id} has no authorization check at all — deliberately. Reading a post is public; the only actions that need protecting are ones that change data. Not every route needs authorization — only apply it where it’s actually meaningful.
update_post and delete_post follow an identical three-step pattern:
1. Fetch the post — if it doesn’t exist, 404. This must happen before any authorization check, otherwise you’d leak information about which IDs exist versus which ones you’re forbidden from touching.
2. Check ownership — if post.user_id != current_user.id and not current_user.is_admin: raise 403. This single line is the entire authorization logic for this program: a regular user can only act on posts where user_id matches their own id; an admin bypasses this check entirely.
3. Only then perform the actual update or delete.
Notice authorization always comes after authentication, but before the actual action. Depends(get_current_user) runs first (answering “who is this?”), and only once that succeeds do we reach the ownership check (answering “are they allowed to do this specific thing?”).
Also notice this program repeats the exact same three lines of ownership-check code in two different routes. That’s a code smell worth noticing — and it’s exactly what Program 2 fixes.

Execution & Testing

uv run fastapi dev app/main.py

Register two users, e.g. alice and bob, and log in as each to get two separate tokens.
As alice, create a post via POST /posts/ — note its id.
As bob (using his own token), try PATCH /posts/{alice's post id} — confirm you get a 403 Forbidden.
As alice, PATCH the same post — confirm it succeeds.
As bob, try DELETE /posts/{alice's post id} — confirm 403 again.
Test the admin override: manually set bob‘s is_admin to 1 directly in the database (sqlite3 app.db "UPDATE users SET is_admin = 1 WHERE username = 'bob';"), log in again as bob to get a fresh token, then retry DELETE /posts/{alice's post id} — confirm it now succeeds, since admins bypass the ownership check.

Program 2: A Reusable Ownership-Checking Dependency

Program Objective

Refactor the repeated “fetch post, check 404, check ownership” logic from Program 1 into a single, reusable FastAPI dependency, so every route that needs it can simply declare it as a parameter — following the same clean-dependency philosophy from Chapters 12 and 16.