Token Refresh

Learn the fundamentals of token refresh in FastAPI.

🎯 Core Concept

Token refresh allows users to obtain new access tokens without re-authenticating. This pattern uses short-lived access tokens and long-lived refresh tokens to balance security and user experience.

📖 What You'll Learn

In this section, you'll understand:

• Short-lived access tokens and long-lived refresh tokens

• How it applies to real-world API development

• Implementing secure token rotation

• Common patterns and best practices

💡 Two-Token Pattern

from datetime import datetime, timedelta
from jose import jwt

# Access token: short lifetime (15 minutes)
# Refresh token: long lifetime (7 days)

def create_tokens(user_id: int):
    access_token_expires = timedelta(minutes=15)
    refresh_token_expires = timedelta(days=7)

    access_token = jwt.encode(
        {
            "sub": user_id,
            "type": "access",
            "exp": datetime.utcnow() + access_token_expires
        },
        SECRET_KEY,
        algorithm=ALGORITHM
    )

    refresh_token = jwt.encode(
        {
            "sub": user_id,
            "type": "refresh",
            "exp": datetime.utcnow() + refresh_token_expires
        },
        SECRET_KEY,
        algorithm=ALGORITHM
    )

    return access_token, refresh_token

@app.post("/token")
async def login(credentials: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(credentials.username, credentials.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    access_token, refresh_token = create_tokens(user.id)
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

🔍 Refresh Token Endpoint

@app.post("/refresh")
async def refresh_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

        # Verify it's a refresh token
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=401, detail="Invalid token type")

        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")

        # Create new access token
        access_token_expires = timedelta(minutes=15)
        new_access_token = jwt.encode(
            {
                "sub": user_id,
                "type": "access",
                "exp": datetime.utcnow() + access_token_expires
            },
            SECRET_KEY,
            algorithm=ALGORITHM
        )

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

    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

💼 Production Token Rotation

class TokenStore:
    """Manage token lifecycle and revocation"""
    def __init__(self):
        self.blacklist = set()
        self.refresh_tokens = {}

    def add_to_blacklist(self, token: str):
        self.blacklist.add(token)

    def store_refresh_token(self, user_id: int, refresh_token: str):
        self.refresh_tokens[user_id] = {
            "token": refresh_token,
            "created_at": datetime.utcnow()
        }

    def is_refresh_token_valid(self, user_id: int, token: str):
        stored = self.refresh_tokens.get(user_id)
        return stored and stored["token"] == token

token_store = TokenStore()

@app.post("/token")
async def login(credentials: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(credentials.username, credentials.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    access_token, refresh_token = create_tokens(user.id)
    token_store.store_refresh_token(user.id, refresh_token)

    logger.info(f"User {user.id} logged in")
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
        "expires_in": 900  # 15 minutes
    }

@app.post("/refresh")
async def refresh_access_token(refresh_token: str):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")

        # Validate refresh token
        if not token_store.is_refresh_token_valid(user_id, refresh_token):
            raise HTTPException(status_code=401, detail="Invalid refresh token")

        # Create new tokens
        new_access, new_refresh = create_tokens(user_id)
        token_store.store_refresh_token(user_id, new_refresh)

        logger.info(f"User {user_id} refreshed token")
        return {
            "access_token": new_access,
            "refresh_token": new_refresh,
            "token_type": "bearer"
        }

    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/logout")
async def logout(user: User = Depends(get_current_user)):
    # Invalidate refresh token
    token_store.refresh_tokens.pop(user.id, None)
    logger.info(f"User {user.id} logged out")
    return {"message": "Logged out successfully"}

How it applies to real-world API development

Token refresh patterns are essential for:

• **Security**: Short-lived access tokens limit damage if compromised

• **User experience**: Users don't need to re-login frequently

• **Token revocation**: Can revoke refresh tokens when needed

• **Compliance**: Meets security standards for token expiration

Common patterns and best practices

• ✅ Use short-lived access tokens (15 minutes)

• ✅ Use longer-lived refresh tokens (7 days)

• ✅ Validate token type (access vs refresh)

• ✅ Store refresh tokens in database

• ✅ Invalidate refresh tokens on logout

• ✅ Implement token rotation on refresh

• ✅ Monitor refresh token usage patterns

• ✅ Use HTTP-only cookies for refresh tokens

🔑 Key Takeaways

• ✅ Understand the purpose of token refresh

• ✅ Know when to apply this pattern

• ✅ Recognize its benefits in real-world scenarios

• ✅ Be prepared to use it in your projects

Ready to explore more? Check out the advanced section for production patterns and edge cases.