如何从FastAPI服务器中获取客户端详情(当前用户详情)?

huangapple go评论65阅读模式
英文:

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

问题

I can help you translate the code portion you provided into Chinese. Here it is:

# 我正在创建一个带有用户管理的应用程序,但需要添加一个路由,该路由将返回当前用户的个人资料详细信息给客户端在 FastAPI 中。
# 但目前我无法编写获取当前用户的功能,尝试从前端获取令牌导致对象未定义。

app = FastAPI()
# 创建 FastAPI 的应用程序对象

engine = create_engine('postgresql://postgres:admin@localhost:5432/postgres')
# 创建用于数据库连接的引擎

SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
# 创建数据库连接的会话

Base = declarative_base()
# 创建用于模型表的 Alchemy Base

class PydanticUsers(BaseModel):
    # 用户输入值的 Pydantic 模式 - 请求体
    username: str
    email: str
    password: str

class PydanticAuth(BaseModel):
    # 用于身份验证的 Pydantic 模型
    username: str
    password: str

class ModelUser(Base):
    # 为模型表创建 Alchemy 类
    __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 ResponseModel1(BaseModel):
    username: str
    email: str

    class Config():
        orm_mode = True

class ResponseModel2(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")

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

Hope this helps! If you have any more specific questions or need further assistance, please feel free to ask.

英文:

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

答案1

得分: 1

为了在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

# 创建FastAPI实例
app = FastAPI()

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

# 数据库引擎和会话设置
engine = create_engine('postgresql://postgres:admin@localhost:5432/postgres')
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

# 创建模型表的基类
Base = declarative_base()

# ... 代码的其余部分 ...

# JWT令牌方案
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

# 依赖项以根据提供的令牌获取当前用户
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

# 路由以检索当前用户的个人资料详细信息
@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=[],  # 在此处添加检索用户博客的逻辑
    )

请确保根据您的项目更新JWT配置或将其删除以使用默认配置。

英文:

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.

huangapple
  • 本文由 发表于 2023年7月13日 15:45:14
  • 转载请务必保留本文链接:https://go.coder-hub.com/76677029.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定