我正在创建一个具有用户管理功能的应用程序,但需要添加一个路由,该路由将向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"}
字符串
1条答案
按热度按时间e5nqia271#
要在FastAPI中获取当前用户,您必须获取JWT令牌并对其进行解码以从中提取信息,希望它有所帮助:
字符串
请确保根据您的项目更新JWT配置,或删除它们以使用默认值。