How to ensure I have a fresh access token with starlette.

[gillespied]

I read this comment: #359
But I'm really struggling to see how to implement it. I include this minimal app below. I want to use cognito to permit access to my app, I am also using cognito to secure an api in aws. At somepoint I want to call the api endpoint so need a new access token. This may be a long time after I'm logged in. I thought I might refresh before calling the token. That let me to the linked comment that says the client handles refreshing automatically but I am at a loss to see how I acheive that.

Could someone please give me a steer what I need to do? I was trying to keep track of the tokens, but I didnt really want to have server side storage, and they are too large so they get silently dropped.

import json
import os
from datetime import datetime

import httpx
import uvicorn
from authlib.integrations.starlette_client import OAuth, OAuthError
from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from starlette.middleware.sessions import SessionMiddleware

import config

app = FastAPI()
# Use a secure random key in production
app.add_middleware(SessionMiddleware, secret_key=os.urandom(24).hex())

# Setup templating and static files
templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")

# Initialize OAuth
oauth = OAuth()

oauth.register(
    name="oidc",
    authority=config.COGNITO_AUTHORITY,
    client_id=config.COGNITO_CLIENT_ID,
    client_secret=config.COGNITO_CLIENT_SECRET,
    server_metadata_url=config.COGNITO_META_DATA_URL,
    client_kwargs={"scope": "email openid profile"},
)


def calculate_token_expiry(expiry_timestamp):
    """Calculate token expiry details"""
    if expiry_timestamp:
        expiry_datetime = datetime.fromtimestamp(expiry_timestamp)
        expiry_str = expiry_datetime.strftime("%Y-%m-%d %H:%M:%S")

        # Calculate seconds remaining
        now = datetime.now()
        seconds_remaining = (expiry_datetime - now).total_seconds()

        return {
            "expiry": expiry_str,
            "seconds_remaining": max(0, int(seconds_remaining)),
        }
    return {"expiry": "Unknown", "seconds_remaining": 0}


@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
    user = request.session.get("user")
    if user:
        # Redirect to dashboard if already logged in
        return RedirectResponse(url="/dashboard")

    return templates.TemplateResponse("index.html", {"request": request})


@app.get("/login")
async def login(request: Request):
    # Build absolute URL for callback
    redirect_uri = request.url_for("auth_callback")
    return await oauth.oidc.authorize_redirect(request, redirect_uri)


@app.get("/auth/callback")
async def auth_callback(request: Request):
    try:
        token = await oauth.oidc.authorize_access_token(request)
        user = token.get("userinfo")
        token["id_token"] = token["id_token"][:100] # trim these tokens else they are too long to store in the cookie. 
        token["access_token"] = token["access_token"][:100]
        token["refresh_token"] = token["refresh_token"][:100]
        if user:
            request.session["user"] = dict(user)
            request.session["token"] = token
        return RedirectResponse(url="/dashboard")
    except OAuthError as error:
        return templates.TemplateResponse(
            "error.html", {"request": request, "error": error.error}
        )


@app.get("/dashboard", response_class=HTMLResponse)
async def dashboard(request: Request):
    print("In dashboard")
    print(request.session)
    user = request.session.get("user")
    token = request.session.get("token")

    if not user:
        return RedirectResponse(url="/")

    # Extract token information
    token_info = {
        "email": user.get("email", "Not available"),
        **calculate_token_expiry(user.get("exp", None)),
    }

    # Format token for display (removing sensitive parts)
    display_token = {}
    if token:
        access_token = token.get("access_token", "")
        display_token = {
            "email": user.get("email", "Not available"),
            "access_token": access_token[:10] + "..." if access_token else "",
            "token_type": user.get("token_use", ""),
        }

    return templates.TemplateResponse(
        "dashboard.html",
        {
            "request": request,
            "user": user,
            "token_info": token_info,
            "display_token": json.dumps(display_token, indent=2),
        },
    )


@app.post("/call-api")
async def call_api(request: Request):
    ## Here i want to call an external api for which i might need to get an access token
    #  How do i ensure I have a fresh one.
    pass


@app.get("/logout")
async def logout(request: Request):
    request.session.pop("user", None)
    request.session.pop("token", None)
    return RedirectResponse(url="/")


if __name__ == "__main__":
    uvicorn.run("main:app", host="localhost", port=3000, reload=True)