How to get client details(Current User Details) from Server in FastAPI?

Question

I'm creating an application with user management , but need to add a route which will return the current user profile details back to the client in FastAPI. But currently I'm unable to code the functionality for getting current user , trying to get the token from front end results in object being undefined.

app = FastAPI()
'''Creates Object of FastAPI'''


engine = create_engine('postgresql://postgres:admin@localhost:5432/postgres')
'''Creates Engine for DB connection'''

SessionLocal=sessionmaker(bind=engine,autocommit=False,autoflush=False,)
'''Creates Session for DB connection'''

Base = declarative_base()
'''Creates Alchemy Base for Model Tables '''

class PydanticUsers(BaseModel):
    '''Pydantic schema for the User to input values-body'''
    username:str
    email:str
    password:str


class PydanticAuth(BaseModel):
    '''Pydantic Model for Authentication'''
    username:str
    password:str




class ModelUser(Base):  
    '''Creates Alchemy Class for Model Tables '''
    __tablename__="users"
    id =Column(Integer,primary_key=True,index=True)
    username = Column(String, unique=True)
    email=Column(String, unique=True)
    password=Column(String)
    admin=Column(Boolean,default=False)

class ResponceModel1(BaseModel):
    username:str
    email:str
    class Config():
        orm_mode =True

class ResponceModel2(BaseModel):
    id:int
    username:str
    email:str
    admin:bool
    blogs:List
    class Config():
        orm_mode =True

Base.metadata.create_all(engine) #Migration
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

def verify_token(token:str,credentials_exception):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception

def get_db():
    db=SessionLocal()
    try:
        yield db 
    finally:
        db.close()
pwd_context=CryptContext(schemes=["bcrypt"], deprecated ="auto")

def get_password(password,hashed_password):
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

class Hash():
    def bcrypt(password: str):
        return pwd_context.hash(password)
    def verify(hashed_password,plain_password):
        return pwd_context.verify(plain_password,hashed_password)

@app.post('/login',tags=['authentication'],status_code=status.HTTP_202_ACCEPTED)
def login(request:OAuth2PasswordRequestForm = Depends(),db: Session = Depends(get_db)):
    print(request.username)
    user=db.query(ModelUser).filter(ModelUser.username ==request.username).first()
    print(user)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,\
                            detail=f"Invalid Credential-User")
    if not Hash.verify(user.password,request.password):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,\
                            detail=f"Invalid Credential-Pass")
    access_token = create_access_token(data={"sub": user.username},)
    return {"access_token": access_token, "token_type": "bearer"}

Answer 1

To get the current user in FastAPI , you have to get JWT token and decode it to extract information from it, hope it helps:

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import List

# Create FastAPI instance
app = FastAPI()

# JWT configuration
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Database engine and session setup
engine = create_engine('postgresql://postgres:admin@localhost:5432/postgres')
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

# Create Base for Model Tables
Base = declarative_base()

# ... Rest of your code ...

# JWT token scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

# Dependency to get the current user based on the provided token
def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid authentication credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        user = db.query(ModelUser).filter(ModelUser.username == username).first()
        if user is None:
            raise credentials_exception
        return user
    except JWTError:
        raise credentials_exception

# Route to retrieve the current user profile details
@app.get('/profile', tags=['user'])
def get_user_profile(current_user: ModelUser = Depends(get_current_user)):
    return ResponceModel2(
        id=current_user.id,
        username=current_user.username,
        email=current_user.email,
        admin=current_user.admin,
        blogs=[],  # Add your logic to retrieve the user's blogs here
    )

Make sure to update JWT configuration based on your project or remove them to use defaults.