oauth2.0 如何在FastAPI中从服务器获取客户端详细信息(当前用户详细信息)?

gojuced7  于 2023-08-02  发布在  其他
关注(0)|答案(1)|浏览(162)

我正在创建一个具有用户管理功能的应用程序,但需要添加一个路由,该路由将向FastAPI中的客户端返回当前用户配置文件的详细信息。但是目前我无法编写获取当前用户的功能,试图从前端获取令牌导致对象未定义。

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"}

字符串

e5nqia27

e5nqia271#

要在FastAPI中获取当前用户,您必须获取JWT令牌并对其进行解码以从中提取信息,希望它有所帮助:

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
    )

字符串
请确保根据您的项目更新JWT配置,或删除它们以使用默认值。

相关问题