FastAPI中实现获取当前用户详情接口的问题求助

现有代码的核心问题 #

你的代码存在几个关键缺失，导致无法正确获取当前用户：

• 未定义TokenData类，但verify_token函数中直接使用了该类
• 缺失JWT认证必需的全局常量：SECRET_KEY、ALGORITHM、ACCESS_TOKEN_EXPIRE_MINUTES
• 没有实现获取当前用户的依赖函数，无法在路由中注入用户对象
• 前端可能未正确在请求头中携带token，导致后端无法识别用户身份

完整修复后的代码 #

补全所有缺失部分，并添加用户资料查询路由：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Optional, List

# JWT核心配置（SECRET_KEY建议用openssl rand -hex 32生成安全密钥）
SECRET_KEY = "your-secure-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

app = FastAPI()
'''FastAPI实例化'''

engine = create_engine('postgresql://postgres:admin@localhost:5432/postgres')
'''数据库连接引擎'''

SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False,)
'''数据库会话工厂'''

Base = declarative_base()
'''SQLAlchemy模型基类'''

class PydanticUsers(BaseModel):
    '''用户创建请求的Pydantic schema'''
    username: str
    email: str
    password: str

class PydanticAuth(BaseModel):
    '''用户认证请求的Pydantic模型'''
    username: str
    password: str

class TokenData(BaseModel):
    '''JWT解析后的数据模型'''
    username: Optional[str] = None

class ModelUser(Base):  
    '''数据库用户表模型'''
    __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)  # 数据库表迁移
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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
        return TokenData(username=username)
    except JWTError:
        raise credentials_exception

def get_db():
    db = SessionLocal()
    try:
        yield db 
    finally:
        db.close()

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta if expires_delta else 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():
    @staticmethod
    def bcrypt(password: str):
        return pwd_context.hash(password)
    @staticmethod
    def verify(hashed_password, plain_password):
        return pwd_context.verify(plain_password, hashed_password)

# 新增：获取当前用户的依赖函数
def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证用户凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    token_data = verify_token(token, credentials_exception)
    user = db.query(ModelUser).filter(ModelUser.username == token_data.username).first()
    if not user:
        raise credentials_exception
    return user

# 新增：获取当前用户资料的路由
@app.get("/users/me", response_model=ResponceModel2, tags=["user"])
def read_users_me(current_user: ModelUser = Depends(get_current_user)):
    return current_user

@app.post('/login', tags=['authentication'], status_code=status.HTTP_202_ACCEPTED)
def login(request: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = db.query(ModelUser).filter(ModelUser.username == request.username).first()
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="无效凭据：用户不存在")
    if not Hash.verify(user.password, request.password):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="无效凭据：密码错误")
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

关键修复点说明 #

1. 补全缺失组件
   • 添加TokenData类，用于解析JWT payload中的用户标识
   • 补充JWT认证必需的全局常量，这是token生成与验证的核心依据
2. 实现用户身份注入
   • 新增get_current_user依赖函数：自动从请求头提取token、验证有效性、查询数据库返回用户对象，可直接在路由中注入使用
3. 添加用户资料路由
   • /users/me路由通过依赖注入获取当前用户，返回完整的用户资料

前端正确携带token的方式 #

前端请求/users/me时，必须在请求头中添加Authorization字段，格式为：

Authorization: Bearer <登录接口获取的access_token>

若前端出现"对象未定义"错误，大概率是未正确设置该请求头，或token已过期/无效。

JavaScript fetch示例：

const accessToken = "从登录接口获取的token字符串";
fetch("http://localhost:8000/users/me", {
  method: "GET",
  headers: {
    "Authorization": `Bearer ${accessToken}`
  }
})
.then(response => response.json())
.then(data => console.log(data));

内容的提问来源于stack exchange，提问作者Junoo Jacob Maliyil