ARTICLE AD BOX
I'm building a login system with FastAPI and fastapi-login. I already have a /register route working.
My goal is:
If a valid token is present, allow access to certain routes.
If no token or an invalid token is present, force a fresh login.
The problem is:
Even when a valid token exists, calling manager.optional(request) always returns None. This breaks my logic because it always triggers a fresh login instead of recognizing the valid token.
My Code:
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status from sqlalchemy.orm import Session from passlib.context import CryptContext from fastapi_login import LoginManager from fastapi.responses import RedirectResponse from app.db.models import User from app.db.database import get_db from app.schemas.user_schema import UserCreate from datetime import timedelta SECRET_KEY = "abcdefghijklmno" TOKEN_URL = "/login" manager = LoginManager(SECRET_KEY, token_url=TOKEN_URL, use_cookie=True) # manager.cookie_name = "access_token" router = APIRouter(prefix="/auth", tags=["Authentication"]) pwd = CryptContext(schemes=["argon2"], deprecated="auto") # user registration @router.post("/register", status_code=status.HTTP_201_CREATED) def register_user(data: UserCreate, response: Response, db: Session = Depends(get_db)): email_normalized = data.email.strip().lower() existing = db.query(User).filter(User.email == email_normalized).first() if existing: raise HTTPException(status_code=400, detail="Email already exists") hashed_pass = pwd.hash(data.password) new_user = User( name=data.name.strip(), email=data.email.strip(), phone=data.phone.strip(), address=data.address.strip(), fitness_goal=data.fitnessGoal.strip(), experience_level=data.experienceLevel.strip(), password=hashed_pass ) db.add(new_user) db.commit() db.refresh(new_user) access_token = manager.create_access_token(data={"sub": new_user.email}) manager.set_cookie(response, access_token) return {"message": "User registered successfully."} # user login @router.post("/login", status_code=status.HTTP_200_OK) async def login(request: Request, response: Response, db: Session = Depends(get_db)): try: # If token valid -> auto login payload = await manager.optional(request) user_email = payload.get('sub') user = db.query(User).filter(User.email == user_email).first() if user: return {"message": "User already logged in"} except Exception as e: pass # token invalid or expired → continue to normal login # Get login body (email & password) try: body = await request.json() except Exception: raise HTTPException(status_code=400, detail="Invalid request body") email = body.get("email", "").strip().lower() password = body.get("password", "") # Validate user user = db.query(User).filter(User.email == email).first() if not user or not pwd.verify(password, user.password): raise HTTPException( status_code=401, detail="Invalid email or password") # Create new token + set cookie access_token = manager.create_access_token(data={"sub": user.email}, expires=timedelta(hours=12)) manager.set_cookie(response, access_token) return {"message": "Login successful"}Expected behavior:
If the token is valid, manager.optional(request) should return the user.
If the token is invalid or missing, it should return None.
Question:
Why does manager.optional(request) always return None even when I send a valid token? How can I make it properly detect a valid token and only force login when needed?
