Authentication with FastAPI

FastAPI provides built-in support for JWT-based authentication. This tutorial covers building a complete auth system with user registration, login, and protected endpoints.

Prerequisites

Install the required dependencies:

pip install fastapi python-jose passlib[bcrypt] python-multipart

• python-jose: JWT token creation and validation
• passlib[bcrypt]: Secure password hashing
• python-multipart: Form data handling for login

User Model with Password Hashing

Create a user model that stores hashed passwords instead of plaintext:

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

app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None

class UserInDB(User):
    hashed_password: str

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

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

# In production, store users in a database
users_db = {}

def create_user(username: str, password: str, email: str = None) -> UserInDB:
    hashed = get_password_hash(password)
    user = UserInDB(username=username, email=email, hashed_password=hashed)
    users_db[username] = user
    return user

# Create a test user
create_user("testuser", "testpass123", "test@example.com")

Key points:

• Never store plaintext passwords — always hash them
• bcrypt adds a salt automatically
• verify_password() safely compares hashes

JWT Token Creation

Create tokens that expire after a set time:

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

@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

The token contains:

• sub: The username (subject)
• exp: Expiration timestamp

Protected Routes

Use dependency injection to protect routes:

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: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = users_db.get(username)
    if user is None:
        raise credentials_exception
    return user

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

The get_current_user dependency:

1. Extracts the token from the Authorization header
2. Decodes and validates the JWT
3. Looks up the user in the database
4. Returns the user or raises 401

Complete Example

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

app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

class User(BaseModel):
    username: str
    email: str | None = None

class UserInDB(User):
    hashed_password: str

users_db = {}

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

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    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: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = users_db.get(username)
    if user is None:
        raise credentials_exception
    return user

@app.post("/register")
def register(form_data: OAuth2PasswordRequestForm = Depends()):
    if form_data.username in users_db:
        raise HTTPException(status_code=400, detail="Username already exists")
    hashed = get_password_hash(form_data.password)
    users_db[form_data.username] = UserInDB(
        username=form_data.username, 
        hashed_password=hashed
    )
    return {"msg": "User created"}

@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Incorrect credentials")
    
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/protected")
def protected_route(current_user: User = Depends(get_current_user)):
    return {"username": current_user.username}

Test it:

# Register a user
curl -X POST http://localhost:8000/register -d "username=alice&password=secret"

# Login to get token
curl -X POST http://localhost:8000/token -d "username=alice&password=secret"
# Returns: {"access_token": "...", "token_type": "bearer"}

# Access protected route
curl -H "Authorization: Bearer YOUR_TOKEN" http://localhost:8000/protected

Security Considerations

• Change the secret key in production — use environment variables
• Use HTTPS in production to protect tokens in transit
• Add token refresh for long-lived sessions
• Implement rate limiting on login endpoints to prevent brute force
• Store tokens securely on the client side (httpOnly cookies recommended)

See Also

• Flask Basics — Previous tutorial in the series
• Django Basics — Previous tutorial in the series
• python-jose documentation — JWT handling