Python源码示例：fastapi.Security()
示例1
def _get_authorization_header(
api_key: str = Security(RWAPIKeyHeader(name=HEADER_KEY)),
) -> str:
try:
token_prefix, token = api_key.split(" ")
except ValueError:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail=strings.WRONG_TOKEN_PREFIX,
)
if token_prefix != JWT_TOKEN_PREFIX:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail=strings.WRONG_TOKEN_PREFIX,
)
return token
示例2
def get_current_user(token: str = Security(reusable_oauth2)):
try:
payload = jwt.decode(token, config.SECRET_KEY, algorithms=[ALGORITHM])
token_data = TokenPayload(**payload)
except PyJWTError:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Could not validate credentials"
)
bucket = get_default_bucket()
user = crud.user.get(bucket, username=token_data.username)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
示例3
def get_current_active_user(current_user: UserInDB = Security(get_current_user)):
if not crud.user.is_active(current_user):
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
示例4
def get_current_active_superuser(current_user: UserInDB = Security(get_current_user)):
if not crud.user.is_superuser(current_user):
raise HTTPException(
status_code=400, detail="The user doesn't have enough privileges"
)
return current_user
示例5
def get_current_user(oauth_header: Optional[str] = Security(oid)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
示例6
def get_current_user(oauth_header: Optional[str] = Security(api_key)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
示例7
def read_user(
user_data: Tuple[str, List[str]] = Security(get_user, scopes=["foo", "bar"]),
data: List[int] = Depends(get_data),
):
return {"user": user_data[0], "scopes": user_data[1], "data": data}
示例8
def get_current_user(oauth_header: Optional[str] = Security(reusable_oauth2)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
示例9
def read_items(token: Optional[str] = Security(oauth2_scheme)):
return {"token": token}
示例10
def read_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Security(security),
):
if credentials is None:
return {"msg": "Create an account first"}
return {"scheme": credentials.scheme, "credentials": credentials.credentials}
示例11
def get_current_user(oauth_header: Optional[str] = Security(api_key)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
示例12
def read_current_user(credentials: HTTPAuthorizationCredentials = Security(security)):
return {"scheme": credentials.scheme, "credentials": credentials.credentials}
示例13
def read_current_user(credentials: HTTPAuthorizationCredentials = Security(security)):
return {"scheme": credentials.scheme, "credentials": credentials.credentials}
示例14
def get_current_user(oauth_header: str = Security(api_key)):
user = User(username=oauth_header)
return user
示例15
def read_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Security(security),
):
if credentials is None:
return {"msg": "Create an account first"}
return {"scheme": credentials.scheme, "credentials": credentials.credentials}
示例16
def read_current_user(credentials: Optional[HTTPBasicCredentials] = Security(security)):
if credentials is None:
return {"msg": "Create an account first"}
return {"username": credentials.username, "password": credentials.password}
示例17
def get_current_user(oauth_header: str = Security(api_key)):
user = User(username=oauth_header)
return user
示例18
def get_current_user(oauth_header: Optional[str] = Security(api_key)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
示例19
def read_current_user(credentials: HTTPBasicCredentials = Security(security)):
return {"username": credentials.username, "password": credentials.password}
示例20
def get_current_user(oauth_header: str = Security(oid)):
user = User(username=oauth_header)
return user
示例21
def read_items(token: Optional[str] = Security(oauth2_scheme)):
if token is None:
return {"msg": "Create an account first"}
return {"token": token}
示例22
def read_current_user(credentials: HTTPAuthorizationCredentials = Security(security)):
return {"scheme": credentials.scheme, "credentials": credentials.credentials}
示例23
def get_current_active_user(
current_user: User = Security(get_current_user, scopes=["me"])
):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
示例24
def read_own_items(
current_user: User = Security(get_current_active_user, scopes=["items"])
):
return [{"item_id": "Foo", "owner": current_user.username}]
示例25
def _get_authorization_header_optional(
authorization: Optional[str] = Security(
RWAPIKeyHeader(name=HEADER_KEY, auto_error=False),
),
) -> str:
if authorization:
return _get_authorization_header(authorization)
return ""
示例26
def get_current_user(
db: Session = Depends(get_db), token: str = Security(reusable_oauth2)
):
try:
payload = jwt.decode(token, config.SECRET_KEY, algorithms=[ALGORITHM])
token_data = TokenPayload(**payload)
except PyJWTError:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Could not validate credentials"
)
user = crud.user.get(db, id=token_data.user_id)
if not user:
raise HTTPException(status_code=400, detail="User not found")
return user
示例27
def get_current_active_user(current_user: User = Security(get_current_user)):
if not crud.user.is_active(current_user):
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
示例28
def get_current_active_superuser(current_user: User = Security(get_current_user)):
if not crud.user.is_superuser(current_user):
raise HTTPException(
status_code=400, detail="The user doesn't have enough privileges"
)
return current_user
