在 fastapi 中使 Depends 可选 python
make Depends optional in fastapi python
我有一个 API 可以为用户和非用户提供图像。图片可以是 public 或私有的。
我的代码
@router.get("/{id}")
def get_resource(id: str, current_user: User = Depends(get_current_user)):
return return_resource(id, current_user)
此代码严格执行授权。我想如果用户没有登录,那么它应该把 None
放在 current_user
中,这样我就可以允许访问 public 图像并限制私有。
其他代码
get_current_user
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
async def get_current_user(required: bool = True, token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
if not required and not token:
return None
return verify_token(token, credentials_exception)
我想将 required
之类的参数发送到 get_current_user
verify_token
def verify_token(token: str, credentials_exception):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("email")
pk: str = payload.get("pk")
if email is None:
raise credentials_exception
token_data = TokenData(
email=email,
pk = pk
)
except JWTError:
raise credentials_exception
return token_data
如果我没理解错的话,您可以使用包装函数将参数传递给嵌套函数。像这样:
def get_current_user(required: bool = True):
async def _get_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
if not required and not token:
return None
return verify_token(token, credentials_exception)
return _get_user
@router.get("/{id}")
def get_resource(id: str, current_user: User = Depends(get_current_user(False))):
return return_resource(id, current_user)
我有一个 API 可以为用户和非用户提供图像。图片可以是 public 或私有的。
我的代码
@router.get("/{id}")
def get_resource(id: str, current_user: User = Depends(get_current_user)):
return return_resource(id, current_user)
此代码严格执行授权。我想如果用户没有登录,那么它应该把 None
放在 current_user
中,这样我就可以允许访问 public 图像并限制私有。
其他代码
get_current_user
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
async def get_current_user(required: bool = True, token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
if not required and not token:
return None
return verify_token(token, credentials_exception)
我想将 required
之类的参数发送到 get_current_user
verify_token
def verify_token(token: str, credentials_exception):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("email")
pk: str = payload.get("pk")
if email is None:
raise credentials_exception
token_data = TokenData(
email=email,
pk = pk
)
except JWTError:
raise credentials_exception
return token_data
如果我没理解错的话,您可以使用包装函数将参数传递给嵌套函数。像这样:
def get_current_user(required: bool = True):
async def _get_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
if not required and not token:
return None
return verify_token(token, credentials_exception)
return _get_user
@router.get("/{id}")
def get_resource(id: str, current_user: User = Depends(get_current_user(False))):
return return_resource(id, current_user)