FastAPI · Lesson 7 of 15
OAuth2 and JWT Authentication
Password flow, hashed passwords, JWT access tokens and a get_current_user dependency.
- Intermediate
- 17 min read
- 3 objectives
Before this lessonLesson 6: Middleware, CORS and Trusted Hosts
What you will learn
- Hash a password
- Issue a JWT
- Protect a route
Your Progress
0 of 15 lessons 0%
- Lessons0 / 15
- Completed0
- Est. time left~ 4 hours
Create a free account to keep your progress on every device.
The usual API pattern is: register with a hashed password, log in to receive a JWT, send Authorization: Bearer <token> on later requests. FastAPI's OAuth2 helpers generate the docs UI for that flow.
Hash passwords
from pwdlib import PasswordHash
pwd = PasswordHash.recommended()
def hash_password(plain: str) -> str:
return pwd.hash(plain)
def verify(plain: str, hashed: str) -> bool:
return pwd.verify(plain, hashed)Issue and read a JWT
from datetime import datetime, timedelta, timezone
import jwt
SECRET = "change-me" # settings.secret_key
ALG = "HS256"
def create_token(user_id: int) -> str:
exp = datetime.now(timezone.utc) + timedelta(minutes=30)
return jwt.encode({"sub": str(user_id), "exp": exp}, SECRET, algorithm=ALG)
def parse_token(token: str) -> int:
payload = jwt.decode(token, SECRET, algorithms=[ALG])
return int(payload["sub"])The dependency
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
oauth2 = OAuth2PasswordBearer(tokenUrl="token")
def get_current_user(token: str = Depends(oauth2), db: Session = Depends(get_db)):
try:
user_id = parse_token(token)
except Exception:
raise HTTPException(401, "Invalid token", headers={"WWW-Authenticate": "Bearer"})
user = db.get(User, user_id)
if not user:
raise HTTPException(401, "Invalid token")
return user
@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
user = db.query(User).filter(User.email == form.username).first()
if not user or not verify(form.password, user.password_hash):
raise HTTPException(400, "Incorrect email or password")
return {"access_token": create_token(user.id), "token_type": "bearer"}
@app.get("/me")
def me(user: User = Depends(get_current_user)):
return {"email": user.email}