英文:
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.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论