How To Add Authentication to FastAPI with OAuth2 and JWT

Intermediate

You have built a FastAPI application with clean endpoints and Pydantic validation. Everything works perfectly — until you realize anyone on the internet can call your API. No login required, no identity checks, no permission controls. This is the exact moment every API developer reaches: your endpoints need authentication, and you need it done properly without building a security framework from scratch.

FastAPI has built-in support for OAuth2 with Password flow and integrates cleanly with JSON Web Tokens (JWT). You will need three packages beyond FastAPI itself: python-jose[cryptography] for creating and verifying JWT tokens, passlib[bcrypt] for secure password hashing, and python-multipart for handling form data in the login endpoint. Install them with pip install python-jose[cryptography] passlib[bcrypt] python-multipart.

This tutorial walks you through the complete authentication flow: hashing and verifying passwords, creating JWT access tokens, protecting endpoints with dependency injection, extracting the current user from tokens, and implementing role-based access control. By the end, you will have a reusable authentication system that you can drop into any FastAPI project.

JWT Authentication in 30 Seconds

Here is the simplest possible protected endpoint in FastAPI. This gives you the core pattern before we build out the full system.

# quick_auth.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

@app.get("/protected")
def protected_route(token: str = Depends(oauth2_scheme)):
    # In a real app, you would decode and verify the JWT here
    if token != "secret-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"message": "You have access!", "token": token}

Output (without token):

{"detail": "Not authenticated"}

Output (with valid token in Authorization header):

{"message": "You have access!", "token": "secret-token"}

The OAuth2PasswordBearer dependency automatically extracts the token from the Authorization: Bearer <token> header. If the header is missing, FastAPI returns a 401 response before your function even runs. The tokenUrl parameter tells Swagger UI where to send login requests.

How JWT Authentication Works

Before writing the full implementation, it helps to understand the flow. JWT (JSON Web Token) authentication works in four steps: the client sends credentials (username and password), the server verifies them and returns a signed token, the client includes that token in every subsequent request, and the server verifies the token signature on each protected endpoint.

Step	Who	What Happens
1. Login	Client	Sends username + password to /login
2. Token creation	Server	Verifies password, creates signed JWT
3. Authenticated request	Client	Sends JWT in Authorization: Bearer header
4. Token verification	Server	Decodes JWT, checks signature and expiry

The JWT itself contains encoded JSON with the user’s identity (called “claims”), an expiration time, and a cryptographic signature. The server signs the token with a secret key, so any tampering is detectable without a database lookup. This makes JWT ideal for stateless APIs where you do not want to store session data on the server.

Password Hashing with Passlib

Never store passwords in plain text. Use passlib with the bcrypt algorithm to hash passwords before storing them and verify them during login.

# password_utils.py
from passlib.context import CryptContext

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

def hash_password(password: str) -> str:
    """Hash a plain text password for storage."""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Check a plain text password against a stored hash."""
    return pwd_context.verify(plain_password, hashed_password)

# Example usage
hashed = hash_password("my_secure_password")
print(f"Hashed: {hashed}")
print(f"Verify correct: {verify_password('my_secure_password', hashed)}")
print(f"Verify wrong: {verify_password('wrong_password', hashed)}")

Output:

Hashed: $2b$12$LJ3m4ys3Lg...Kz8dHJKe (unique each time)
Verify correct: True
Verify wrong: False

The CryptContext handles algorithm selection, salt generation, and hash verification. The deprecated="auto" setting means passlib will automatically upgrade old hashes to the current scheme when passwords are verified. Each hash is unique even for the same password because bcrypt generates a random salt internally.

Creating and Decoding JWT Tokens

The python-jose library creates and verifies JWT tokens. Each token contains a payload (claims) signed with your secret key.

# jwt_utils.py
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError

SECRET_KEY = "your-secret-key-keep-this-safe-and-long"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    """Create a JWT access token with an expiration time."""
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_access_token(token: str) -> dict:
    """Decode and verify a JWT token. Raises JWTError if invalid."""
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

# Example
token = create_access_token({"sub": "alice", "role": "admin"})
print(f"Token: {token[:50]}...")

payload = decode_access_token(token)
print(f"Payload: {payload}")

Output:

Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzd...
Payload: {'sub': 'alice', 'role': 'admin', 'exp': 1712534400}

The sub (subject) claim is a JWT standard for identifying the user. The exp claim sets when the token expires — after this time, decode_access_token raises a JWTError automatically. In production, store SECRET_KEY in an environment variable, not in your source code.

Building the Complete Authentication System

Now let us combine password hashing and JWT tokens into a complete FastAPI authentication system with a user database, login endpoint, and protected routes.

# auth_app.py
from datetime import datetime, timedelta, timezone
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from pydantic import BaseModel

# Configuration
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

app = FastAPI(title="Auth Demo")

# Simulated user database
fake_users_db = {
    "alice": {
        "username": "alice",
        "hashed_password": pwd_context.hash("alice123"),
        "role": "admin",
    },
    "bob": {
        "username": "bob",
        "hashed_password": pwd_context.hash("bob456"),
        "role": "user",
    },
}

# Pydantic models
class Token(BaseModel):
    access_token: str
    token_type: str

class User(BaseModel):
    username: str
    role: str

# Helper functions
def authenticate_user(username: str, password: str):
    user = fake_users_db.get(username)
    if not user or not pwd_context.verify(password, user["hashed_password"]):
        return None
    return user

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return User(username=user["username"], role=user["role"])

# Endpoints
@app.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token(data={"sub": user["username"], "role": user["role"]})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/me", response_model=User)
async def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user

@app.get("/admin")
async def admin_only(current_user: User = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return {"message": f"Welcome admin {current_user.username}!"}

@app.get("/public")
async def public_endpoint():
    return {"message": "This endpoint is open to everyone"}

Testing the login flow:

# Step 1: Login to get a token
curl -X POST http://127.0.0.1:8000/login \
  -d "username=alice&password=alice123"

# Response:
{"access_token": "eyJhbGciOiJIUzI1NiI...", "token_type": "bearer"}

# Step 2: Access protected endpoint with the token
curl http://127.0.0.1:8000/me \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiI..."

# Response:
{"username": "alice", "role": "admin"}

# Step 3: Access admin-only endpoint
curl http://127.0.0.1:8000/admin \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiI..."

# Response:
{"message": "Welcome admin alice!"}

The get_current_user dependency is the heart of this system. It extracts the token from the header, decodes it, looks up the user, and returns a User object — all before your endpoint function runs. Any endpoint that includes current_user: User = Depends(get_current_user) is automatically protected.

Role-Based Access Control

The admin endpoint above uses a simple role check, but you can make this reusable with a dependency factory that creates role-checking dependencies on the fly.

# role_checker.py
from fastapi import Depends, HTTPException, status

def require_role(required_role: str):
    """Create a dependency that checks if the user has the required role."""
    async def role_checker(current_user: User = Depends(get_current_user)):
        if current_user.role != required_role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Role '{required_role}' required. You have '{current_user.role}'.",
            )
        return current_user
    return role_checker

# Usage in endpoints
@app.get("/admin/dashboard")
async def admin_dashboard(admin: User = Depends(require_role("admin"))):
    return {"message": "Admin dashboard", "user": admin.username}

@app.get("/editor/publish")
async def editor_publish(editor: User = Depends(require_role("editor"))):
    return {"message": "Editor publish page", "user": editor.username}

Output (bob tries /admin/dashboard):

{"detail": "Role 'admin' required. You have 'user'."}

The require_role function returns a new dependency for each role. This pattern scales cleanly — add as many roles as your application needs without duplicating validation logic in every endpoint.

Token Refresh Strategy

Access tokens should be short-lived (15-30 minutes) for security. But you do not want users logging in every 30 minutes. The standard solution is a refresh token with a longer lifespan that can generate new access tokens.

# token_refresh.py
REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_refresh_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/login", response_model=dict)
async def login_with_refresh(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    access_token = create_access_token({"sub": user["username"], "role": user["role"]})
    refresh_token = create_refresh_token({"sub": user["username"]})
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
    }

@app.post("/refresh", response_model=Token)
async def refresh_access_token(refresh_token: str):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=401, detail="Invalid token type")
        username = payload.get("sub")
        new_token = create_access_token({"sub": username, "role": fake_users_db[username]["role"]})
        return {"access_token": new_token, "token_type": "bearer"}
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

The refresh token has a type claim set to "refresh" so it cannot be used as an access token. When the access token expires, the client sends the refresh token to /refresh to get a new access token without re-entering credentials.

Security Best Practices

Authentication code is one place where shortcuts cause real damage. Here are the practices that matter most for a production FastAPI application.

Practice	Why It Matters
Store SECRET_KEY in env vars	Keys in source code end up in Git history
Use bcrypt (not MD5/SHA)	Bcrypt is intentionally slow, resistant to brute force
Set short token expiry (15-30 min)	Limits damage if a token is stolen
Use HTTPS in production	Tokens sent over HTTP can be intercepted
Validate token claims (sub, exp)	Missing validation lets forged tokens through
Return generic error messages	“Invalid credentials” not “User not found” prevents user enumeration

Real-Life Example: Protected Notes API

Let us build a practical application: a notes API where each user can only see and modify their own notes. This demonstrates how authentication integrates with real CRUD operations.

# notes_api.py
from datetime import datetime, timedelta, timezone
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from pydantic import BaseModel
from typing import Optional

SECRET_KEY = "notes-app-secret-change-me"
ALGORITHM = "HS256"
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

app = FastAPI(title="Protected Notes API")

# Databases
users_db = {
    "alice": {"username": "alice", "hashed_password": pwd_context.hash("alice123")},
    "bob": {"username": "bob", "hashed_password": pwd_context.hash("bob456")},
}
notes_db = {}
next_note_id = 1

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

class NoteCreate(BaseModel):
    title: str
    content: str

class NoteResponse(BaseModel):
    id: int
    title: str
    content: str
    owner: str
    created_at: str

# Auth helpers
def create_token(username: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=30)
    return jwt.encode({"sub": username, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None or username not in users_db:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

# Endpoints
@app.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = users_db.get(form_data.username)
    if not user or not pwd_context.verify(form_data.password, user["hashed_password"]):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    return {"access_token": create_token(form_data.username), "token_type": "bearer"}

@app.post("/notes", response_model=NoteResponse, status_code=201)
async def create_note(note: NoteCreate, username: str = Depends(get_current_user)):
    global next_note_id
    note_data = {
        "id": next_note_id,
        "title": note.title,
        "content": note.content,
        "owner": username,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    notes_db[next_note_id] = note_data
    next_note_id += 1
    return note_data

@app.get("/notes", response_model=list[NoteResponse])
async def list_my_notes(username: str = Depends(get_current_user)):
    return [n for n in notes_db.values() if n["owner"] == username]

@app.delete("/notes/{note_id}", status_code=204)
async def delete_note(note_id: int, username: str = Depends(get_current_user)):
    note = notes_db.get(note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    if note["owner"] != username:
        raise HTTPException(status_code=403, detail="Not your note")
    del notes_db[note_id]

Testing the flow:

# Login as Alice
curl -X POST http://127.0.0.1:8000/login -d "username=alice&password=alice123"
# {"access_token": "eyJ...", "token_type": "bearer"}

# Create a note (use Alice's token)
curl -X POST http://127.0.0.1:8000/notes \
  -H "Authorization: Bearer eyJ..." \
  -H "Content-Type: application/json" \
  -d '{"title": "Shopping List", "content": "Milk, eggs, bread"}'
# {"id": 1, "title": "Shopping List", "content": "Milk, eggs, bread", "owner": "alice", ...}

# Bob cannot see Alice's notes
curl http://127.0.0.1:8000/notes -H "Authorization: Bearer BOB_TOKEN"
# [] (empty list -- Bob has no notes)

Each note is tagged with its owner, and the list_my_notes endpoint filters by the authenticated user. The delete endpoint checks ownership before allowing deletion. This pattern of tying data to the authenticated user is fundamental to building multi-tenant APIs.

Frequently Asked Questions

Should I use sessions or JWT for my API?

Use JWT for stateless APIs (especially mobile or SPA clients) where you do not want to maintain server-side session storage. Use sessions for traditional server-rendered web applications where the backend controls the full page lifecycle. JWT works well for microservices because any service can verify the token independently without a shared session store.

How should I generate and store the SECRET_KEY?

Generate a random key with openssl rand -hex 32 in your terminal. Store it in an environment variable and read it with os.environ["SECRET_KEY"]. Never commit it to version control. In production, use a secrets manager like AWS Secrets Manager, HashiCorp Vault, or your platform’s built-in secrets.

How do I implement password reset?

Create a /forgot-password endpoint that generates a short-lived token (10-15 minutes) and sends it to the user’s email. Create a /reset-password endpoint that accepts the token and the new password. Use the same JWT mechanism but with a different token type claim so reset tokens cannot be used for API access.

Can I add Google or GitHub login?

Yes. Use the authlib or httpx-oauth library to implement OAuth2 authorization code flow. FastAPI’s dependency injection makes it straightforward to add multiple authentication methods. The user logs in through the provider’s OAuth flow, and your server exchanges the authorization code for user profile data.

How do I test authenticated endpoints?

Use FastAPI’s TestClient with the headers parameter. Create a test user, generate a token in your test setup, and include it in requests: client.get("/me", headers={"Authorization": f"Bearer {token}"}). For dependency overrides, use app.dependency_overrides[get_current_user] = lambda: test_user.

Conclusion

FastAPI’s dependency injection system makes OAuth2 and JWT authentication clean and reusable. The get_current_user dependency is the single point of authentication that you inject into any endpoint that needs protection. Combined with passlib for secure password hashing and python-jose for token management, you have a production-ready auth system in under 100 lines of code.

Start with the Protected Notes API example and extend it with a real database, email verification, and OAuth2 social login. The official FastAPI security documentation at fastapi.tiangolo.com/tutorial/security covers advanced patterns including scopes and multiple authentication schemes.

Related Articles

• How To Build a REST API with FastAPI in Python
• FastAPI vs Flask: Which Python Framework Should You Choose?