Python for PHP Developers, Part 5: Refresh Tokens in FastAPI

This is Part 5 of a series. Catch up on Part 1: A Practical Crash Course, Part 2: FastAPI with a Real Database, Part 3: JWT Authentication, and Part 4: Testing FastAPI with pytest.

In Part 3, we issued a 30-minute access token. That's good for security, but users get logged out every half hour. I also mentioned that JWTs aren't revocable by default. This time, we fix both problems with refresh tokens.

1. Why two tokens?

The access token stays stateless and fast. The refresh token is the one we keep track of, so we can cancel it.

2. Update the token helpers

We make two changes to auth.py from Part 3: a type claim, so a refresh token can never be used as an access token, and a new function that creates refresh tokens with a unique ID (jti).

# auth.py (updated)
import uuid
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError

SECRET_KEY = "change-this-in-production"   # load from env, never hardcode
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_access_token(data: dict, expires_delta: timedelta | None = 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, "type": "access"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(username: str):
    jti = str(uuid.uuid4())
    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    payload = {"sub": username, "jti": jti, "type": "refresh", "exp": expire}
    token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return token, jti, expire

def decode_token(token: str) -> dict:
    # Verifies signature and expiry. The "type" claim is checked by the caller.
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        raise ValueError("Invalid or expired token")

I renamed decode_access_token to decode_token, since it now decodes both kinds.

3. Store refresh tokens server-side

# models.py (addition)
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from database import Base

class RefreshToken(Base):
    __tablename__ = "refresh_tokens"

    id = Column(Integer, primary_key=True, index=True)
    jti = Column(String, unique=True, index=True)
    username = Column(String, index=True)
    expires_at = Column(DateTime)
    revoked = Column(Boolean, default=False)

We store only the jti, not the token itself. A leaked database row can't be turned into a working token, because the signature lives in the token, not in the row.

# schemas.py (addition)
class RefreshRequest(BaseModel):
    refresh_token: str

Send the refresh token in the request body, not as a query parameter. Query strings end up in server logs.

4. Login returns both tokens

# main.py (updated login)
from auth import create_access_token, create_refresh_token, decode_token

@app.post("/login")
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.username == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Incorrect username or password")

    access_token = create_access_token({"sub": user.username})
    refresh_token, jti, expires_at = create_refresh_token(user.username)
    db.add(models.RefreshToken(jti=jti, username=user.username, expires_at=expires_at))
    db.commit()

    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
    }

5. The /refresh endpoint: rotation and reuse detection

Every time a refresh token is used, we revoke it and issue a brand new pair. This is called rotation. If an already-used token ever shows up again, something is wrong (it may have been stolen), so we revoke all of that user's tokens.

@app.post("/refresh")
def refresh(body: schemas.RefreshRequest, db: Session = Depends(get_db)):
    try:
        payload = decode_token(body.refresh_token)
    except ValueError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    if payload.get("type") != "refresh":
        raise HTTPException(status_code=401, detail="Wrong token type")

    stored = (
        db.query(models.RefreshToken)
        .filter(models.RefreshToken.jti == payload["jti"])
        .first()
    )
    if not stored:
        raise HTTPException(status_code=401, detail="Unknown refresh token")

    if stored.revoked:
        # A used token came back: possible theft. Revoke everything for this user.
        db.query(models.RefreshToken).filter(
            models.RefreshToken.username == stored.username
        ).update({"revoked": True})
        db.commit()
        raise HTTPException(status_code=401, detail="Refresh token reuse detected")

    # Rotate: revoke the old token, issue a new pair
    stored.revoked = True
    new_refresh, new_jti, expires_at = create_refresh_token(payload["sub"])
    db.add(models.RefreshToken(jti=new_jti, username=payload["sub"], expires_at=expires_at))
    db.commit()

    return {
        "access_token": create_access_token({"sub": payload["sub"]}),
        "refresh_token": new_refresh,
        "token_type": "bearer",
    }

6. Logout that actually works

@app.post("/logout")
def logout(body: schemas.RefreshRequest, db: Session = Depends(get_db)):
    try:
        payload = decode_token(body.refresh_token)
    except ValueError:
        return {"message": "Logged out"}

    stored = (
        db.query(models.RefreshToken)
        .filter(models.RefreshToken.jti == payload.get("jti"))
        .first()
    )
    if stored:
        stored.revoked = True
        db.commit()
    return {"message": "Logged out"}

After logout, the refresh token is dead, so the session can't be renewed.

7. Lock down the access check

Update get_current_user from Part 3 so it rejects refresh tokens:

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = decode_token(token)
    except ValueError:
        raise HTTPException(status_code=401, detail="Could not validate credentials")

    if payload.get("type") != "access":
        raise HTTPException(status_code=401, detail="Wrong token type")
    return payload["sub"]

8. Test it

This builds on the client fixture from Part 4.

# tests/test_refresh.py
from datetime import timedelta
from auth import create_access_token

def login_tokens(client):
    client.post("/register", params={"username": "himani", "password": "secret123"})
    response = client.post(
        "/login", data={"username": "himani", "password": "secret123"}
    )
    return response.json()

def test_refresh_returns_new_tokens(client):
    tokens = login_tokens(client)
    response = client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    assert response.status_code == 200
    assert response.json()["refresh_token"] != tokens["refresh_token"]

def test_refresh_token_cannot_be_reused(client):
    tokens = login_tokens(client)
    client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    second = client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    assert second.status_code == 401

def test_refresh_token_rejected_as_access_token(client):
    tokens = login_tokens(client)
    response = client.post(
        "/items",
        json={"name": "Pen", "price": 1.5},
        headers={"Authorization": f"Bearer {tokens['refresh_token']}"},
    )
    assert response.status_code == 401

def test_expired_access_token_rejected(client):
    expired = create_access_token({"sub": "himani"}, timedelta(minutes=-1))
    response = client.post(
        "/items",
        json={"name": "Pen", "price": 1.5},
        headers={"Authorization": f"Bearer {expired}"},
    )
    assert response.status_code == 401

The expires_delta parameter we added earlier makes the expiry test easy: we create a token that expired a minute ago instead of waiting 30 minutes.

9. Gotchas for PHP developers

• Access tokens still can't be revoked instantly. Refresh tokens fix the long-lived session, but a stolen access token stays valid until it expires. That's why we keep it short.
• Always check the type claim. Without it, a long-lived refresh token could be used as an access token and bypass the whole design.
• Rotation plus reuse detection is the safety net. A stolen refresh token only works once, and the legitimate user's next refresh exposes the theft.
• Where the client stores tokens matters. For browser apps, avoid localStorage for refresh tokens, since any script injected into the page can read it. An HttpOnly cookie is the usual choice.
• Clean up old rows. Expired and revoked tokens pile up in the table. A scheduled job (a cron task, if you're coming from PHP) that deletes expired rows keeps it small.

What's next

The API now has login, refresh, logout, and tests for all of it. Natural next steps are role-based permissions, rate limiting on /login, and running the tests automatically in CI.

How does your team handle token refresh? Let me know in the comments.