Python源码示例：fastapi.Depends()
示例1
def search_users(
q: str,
skip: int = 0,
limit: int = 100,
current_user: UserInDB = Depends(get_current_active_superuser),
):
"""
Search users, use Bleve Query String syntax:
http://blevesearch.com/docs/Query-String-Query/
For typeahead suffix with `*`. For example, a query with: `email:johnd*` will match
users with email `johndoe@example.com`, `johndid@example.net`, etc.
"""
bucket = get_default_bucket()
users = crud.user.search(bucket=bucket, query_string=q, skip=skip, limit=limit)
return users
示例2
def create_user(
*,
user_in: UserCreate,
current_user: UserInDB = Depends(get_current_active_superuser),
):
"""
Create new user.
"""
bucket = get_default_bucket()
user = crud.user.get(bucket, username=user_in.username)
if user:
raise HTTPException(
status_code=400,
detail="The user with this username already exists in the system.",
)
user = crud.user.upsert(bucket, user_in=user_in, persist_to=1)
if config.EMAILS_ENABLED and user_in.email:
send_new_account_email(
email_to=user_in.email, username=user_in.username, password=user_in.password
)
return user
示例3
def update_user_me(
*,
password: str = Body(None),
full_name: str = Body(None),
email: EmailStr = Body(None),
current_user: UserInDB = Depends(get_current_active_user),
):
"""
Update own user.
"""
user_in = UserUpdate(**current_user.dict())
if password is not None:
user_in.password = password
if full_name is not None:
user_in.full_name = full_name
if email is not None:
user_in.email = email
bucket = get_default_bucket()
user = crud.user.update(bucket, username=current_user.username, user_in=user_in)
return user
示例4
def update_user(
*,
username: str,
user_in: UserUpdate,
current_user: UserInDB = Depends(get_current_active_superuser),
):
"""
Update a user.
"""
bucket = get_default_bucket()
user = crud.user.get(bucket, username=username)
if not user:
raise HTTPException(
status_code=404,
detail="The user with this username does not exist in the system",
)
user = crud.user.update(bucket, username=username, user_in=user_in)
return user
示例5
def read_items(
skip: int = 0,
limit: int = 100,
current_user: UserInDB = Depends(get_current_active_user),
):
"""
Retrieve items.
If superuser, all the items.
If normal user, the items owned by this user.
"""
bucket = get_default_bucket()
if crud.user.is_superuser(current_user):
docs = crud.item.get_multi(bucket, skip=skip, limit=limit)
else:
docs = crud.item.get_multi_by_owner(
bucket=bucket, owner_username=current_user.username, skip=skip, limit=limit
)
return docs
示例6
def search_items(
q: str,
skip: int = 0,
limit: int = 100,
current_user: UserInDB = Depends(get_current_active_user),
):
"""
Search items, use Bleve Query String syntax:
http://blevesearch.com/docs/Query-String-Query/
For typeahead suffix with `*`. For example, a query with: `title:foo*` will match
items containing `football`, `fool proof`, etc.
"""
bucket = get_default_bucket()
if crud.user.is_superuser(current_user):
docs = crud.item.search(bucket=bucket, query_string=q, skip=skip, limit=limit)
else:
docs = crud.item.search_with_owner(
bucket=bucket,
query_string=q,
username=current_user.username,
skip=skip,
limit=limit,
)
return docs
示例7
def login(form_data: OAuth2PasswordRequestForm = Depends()):
"""
OAuth2 compatible token login, get an access token for future requests.
"""
bucket = get_default_bucket()
user = crud.user.authenticate(
bucket, username=form_data.username, password=form_data.password
)
if not user:
raise HTTPException(status_code=400, detail="Incorrect email or password")
elif not crud.user.is_active(user):
raise HTTPException(status_code=400, detail="Inactive user")
access_token_expires = timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
data={"username": user.username}, expires_delta=access_token_expires
),
"token_type": "bearer",
}
示例8
def read_users(
skip: int = 0,
limit: int = 100,
current_user: UserInDB = Depends(get_current_active_superuser),
):
"""
Retrieve users.
"""
bucket = get_default_bucket()
users = crud.user.get_multi(bucket, skip=skip, limit=limit)
return users
示例9
def read_user_me(current_user: UserInDB = Depends(get_current_active_user)):
"""
Get current user.
"""
return current_user
示例10
def read_roles(current_user: UserInDB = Depends(get_current_active_superuser)):
"""
Retrieve roles.
"""
roles = crud.utils.ensure_enums_to_strs(RoleEnum)
return {"roles": roles}
示例11
def create_item(
*, item_in: ItemCreate, current_user: UserInDB = Depends(get_current_active_user)
):
"""
Create new item.
"""
bucket = get_default_bucket()
id = crud.utils.generate_new_id()
doc = crud.item.upsert(
bucket=bucket, id=id, doc_in=item_in, owner_username=current_user.username
)
return doc
示例12
def read_item(id: str, current_user: UserInDB = Depends(get_current_active_user)):
"""
Get item by ID.
"""
bucket = get_default_bucket()
doc = crud.item.get(bucket=bucket, id=id)
if not doc:
raise HTTPException(status_code=404, detail="Item not found")
if not crud.user.is_superuser(current_user) and (
doc.owner_username != current_user.username
):
raise HTTPException(status_code=400, detail="Not enough permissions")
return doc
示例13
def delete_item(id: str, current_user: UserInDB = Depends(get_current_active_user)):
"""
Delete an item by ID.
"""
bucket = get_default_bucket()
doc = crud.item.get(bucket=bucket, id=id)
if not doc:
raise HTTPException(status_code=404, detail="Item not found")
if not crud.user.is_superuser(current_user) and (
doc.owner_username != current_user.username
):
raise HTTPException(status_code=400, detail="Not enough permissions")
doc = crud.item.remove(bucket=bucket, id=id)
return doc
示例14
def test_celery(
msg: Msg, current_user: UserInDB = Depends(get_current_active_superuser)
):
"""
Test Celery worker.
"""
celery_app.send_task("app.worker.test_celery", args=[msg.msg])
return {"msg": "Word received"}
示例15
def test_email(
email_to: EmailStr, current_user: UserInDB = Depends(get_current_active_superuser)
):
"""
Test emails.
"""
send_test_email(email_to=email_to)
return {"msg": "Test email sent"}
示例16
def main_depends(commons: dict = Depends(common_parameters)):
return {"in": "main-depends", "params": commons}
示例17
def router_depends(commons: dict = Depends(common_parameters)):
return {"in": "router-depends", "params": commons}
示例18
def overrider_dependency_with_sub(msg: dict = Depends(overrider_sub_dependency)):
return msg
示例19
def read_current_user(current_user: Optional[User] = Depends(get_current_user)):
if current_user is None:
return {"msg": "Create an account first"}
return current_user
示例20
def read_current_user(current_user: Optional[User] = Depends(get_current_user)):
if current_user is None:
return {"msg": "Create an account first"}
return current_user
示例21
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}
示例22
def read_current_user(form_data: OAuth2PasswordRequestFormStrict = Depends()):
return form_data
示例23
def read_current_user(current_user: Optional[User] = Depends(get_current_user)):
if current_user is None:
return {"msg": "Create an account first"}
return current_user
示例24
def get_callable_dependency(value: str = Depends(callable_dependency)):
return value
示例25
def get_callable_gen_dependency(value: str = Depends(callable_gen_dependency)):
return value
示例26
def get_callable_gen_dependency(
value: str = Depends(async_callable_gen_dependency),
):
return value
示例27
def get_synchronous_method_dependency(
value: str = Depends(methods_dependency.synchronous),
):
return value
示例28
def get_synchronous_method_gen_dependency(
value: str = Depends(methods_dependency.synchronous_gen),
):
return value
示例29
def get_asynchronous_method_dependency(
value: str = Depends(methods_dependency.asynchronous),
):
return value
示例30
def get_asynchronous_method_gen_dependency(
value: str = Depends(methods_dependency.asynchronous_gen),
):
return value
