Python源码示例：fastapi.HTTPException()
示例1
def test_wsgi_app(testdir, cli):
module = testdir.make_importable_pyfile(
location="""
from fastapi import FastAPI
from fastapi import HTTPException
app = FastAPI()
@app.get("/api/success")
async def success():
return {"success": True}
@app.get("/api/failure")
async def failure():
raise HTTPException(status_code=500)
return {"failure": True}
"""
)
result = cli.run("/openapi.json", "--app", f"{module.purebasename}:app")
assert result.exit_code == ExitCode.TESTS_FAILED, result.stdout
assert "1 passed, 1 failed in" in result.stdout
示例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(
*,
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
示例4
def update_item(
*,
id: str,
item_in: ItemUpdate,
current_user: UserInDB = Depends(get_current_active_user),
):
"""
Update an item.
"""
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.update(
bucket=bucket, id=id, doc_in=item_in, owner_username=doc.owner_username
)
return doc
示例5
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",
}
示例6
def recover_password(username: str):
"""
Password Recovery.
"""
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.",
)
password_reset_token = generate_password_reset_token(username=username)
send_reset_password_email(
email_to=user.email, username=username, token=password_reset_token
)
return {"msg": "Password recovery email sent"}
示例7
def reset_password(token: str = Body(...), new_password: str = Body(...)):
"""
Reset password.
"""
username = verify_password_reset_token(token)
if not username:
raise HTTPException(status_code=400, detail="Invalid token")
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.",
)
elif not crud.user.is_active(user):
raise HTTPException(status_code=400, detail="Inactive user")
user_in = UserUpdate(name=username, password=new_password)
user = crud.user.update(bucket, username=username, user_in=user_in)
return {"msg": "Password updated successfully"}
示例8
def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate 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
token_data = TokenData(username=username)
except PyJWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user
示例9
def get_closest_place(lat: float, lon: float, es=None):
if es is None:
es = get_elasticsearch()
es_addr = fetch_closest(lat, lon, es=es, max_distance=MAX_DISTANCE_IN_METERS)
places = {
"addr": Address,
"street": Street,
}
loader = places.get(es_addr.get("_type"))
if loader is None:
logger.warning("Found a place with the wrong type")
prometheus.exception("FoundPlaceWithWrongType")
raise HTTPException(
status_code=404,
detail="Closest address to '{}:{}' has a wrong type: '{}'".format(
lat, lon, es_addr.get("_type")
),
)
return loader(es_addr["_source"])
示例10
def get_directions_with_coordinates(
# URL values
f_lon: float,
f_lat: float,
t_lon: float,
t_lat: float,
# Query parameters
type: str,
language: str = "en",
# Request
request: Request = Depends(directions_request),
):
from_place = Latlon(f_lat, f_lon)
to_place = Latlon(t_lat, t_lon)
if not type:
raise HTTPException(status_code=400, detail='"type" query param is required')
return directions_client.get_directions(
from_place, to_place, type, language, params=request.query_params
)
示例11
def raw_autocomplete(self, params, body=None):
url = settings["BRAGI_BASE_URL"] + "/autocomplete"
if body:
response = await self.client.post(url, params=params, json=body)
else:
response = await self.client.get(url, params=params)
if response.status_code != httpx.codes.ok:
try:
explain = response.json()["long"]
except (IndexError, JSONDecodeError):
explain = response.text
logger.error(
'Request to Bragi returned with unexpected status %d: "%s"',
response.status_code,
explain,
)
raise HTTPException(503, "Unexpected geocoder error")
try:
return response.json()
except (JSONDecodeError, pydantic.ValidationError) as e:
logger.exception("Autocomplete invalid response")
raise HTTPException(503, "Invalid response from the geocoder")
示例12
def follow_for_user(
profile: Profile = Depends(get_profile_by_username_from_path),
user: User = Depends(get_current_user_authorizer()),
profiles_repo: ProfilesRepository = Depends(get_repository(ProfilesRepository)),
) -> ProfileInResponse:
if user.username == profile.username:
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST, detail=strings.UNABLE_TO_FOLLOW_YOURSELF,
)
if profile.following:
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST, detail=strings.USER_IS_ALREADY_FOLLOWED,
)
await profiles_repo.add_user_into_followers(
target_user=profile, requested_user=user,
)
return ProfileInResponse(profile=profile.copy(update={"following": True}))
示例13
def unsubscribe_from_user(
profile: Profile = Depends(get_profile_by_username_from_path),
user: User = Depends(get_current_user_authorizer()),
profiles_repo: ProfilesRepository = Depends(get_repository(ProfilesRepository)),
) -> ProfileInResponse:
if user.username == profile.username:
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST,
detail=strings.UNABLE_TO_UNSUBSCRIBE_FROM_YOURSELF,
)
if not profile.following:
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST, detail=strings.USER_IS_NOT_FOLLOWED,
)
await profiles_repo.remove_user_from_followers(
target_user=profile, requested_user=user,
)
return ProfileInResponse(profile=profile.copy(update={"following": False}))
示例14
def create_new_article(
article_create: ArticleInCreate = Body(..., embed=True, alias="article"),
user: User = Depends(get_current_user_authorizer()),
articles_repo: ArticlesRepository = Depends(get_repository(ArticlesRepository)),
) -> ArticleInResponse:
slug = get_slug_for_article(article_create.title)
if await check_article_exists(articles_repo, slug):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=strings.ARTICLE_ALREADY_EXISTS,
)
article = await articles_repo.create_article(
slug=slug,
title=article_create.title,
description=article_create.description,
body=article_create.body,
author=user,
tags=article_create.tags,
)
return ArticleInResponse(article=ArticleForResponse.from_orm(article))
示例15
def mark_article_as_favorite(
article: Article = Depends(get_article_by_slug_from_path),
user: User = Depends(get_current_user_authorizer()),
articles_repo: ArticlesRepository = Depends(get_repository(ArticlesRepository)),
) -> ArticleInResponse:
if not article.favorited:
await articles_repo.add_article_into_favorites(article=article, user=user)
return ArticleInResponse(
article=ArticleForResponse.from_orm(
article.copy(
update={
"favorited": True,
"favorites_count": article.favorites_count + 1,
},
),
),
)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=strings.ARTICLE_IS_ALREADY_FAVORITED,
)
示例16
def get_comment_by_id_from_path(
comment_id: int = Path(..., ge=1),
article: Article = Depends(articles.get_article_by_slug_from_path),
user: Optional[User] = Depends(
authentication.get_current_user_authorizer(required=False),
),
comments_repo: CommentsRepository = Depends(
database.get_repository(CommentsRepository),
),
) -> Comment:
try:
return await comments_repo.get_comment_by_id(
comment_id=comment_id, article=article, user=user,
)
except EntityDoesNotExist:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=strings.COMMENT_DOES_NOT_EXIST,
)
示例17
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
示例18
def _get_current_user(
users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
token: str = Depends(_get_authorization_header_retriever()),
) -> User:
try:
username = jwt.get_username_from_token(token, str(SECRET_KEY))
except ValueError:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail=strings.MALFORMED_PAYLOAD,
)
try:
return await users_repo.get_user_by_username(username=username)
except EntityDoesNotExist:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail=strings.MALFORMED_PAYLOAD,
)
示例19
def test_cli_run_output_success(testdir, cli, workers):
module = testdir.make_importable_pyfile(
location="""
from fastapi import FastAPI
from fastapi import HTTPException
app = FastAPI()
@app.get("/api/success")
async def success():
return {"success": True}
"""
)
result = cli.run(
"/openapi.json", "--app", f"{module.purebasename}:app", f"--workers={workers}", "--show-errors-tracebacks"
)
assert result.exit_code == ExitCode.OK, result.stdout
lines = result.stdout.split("\n")
assert lines[7] == f"Workers: {workers}"
if workers == 1:
assert lines[10].startswith("GET /api/success .")
else:
assert lines[10] == "."
assert " HYPOTHESIS OUTPUT " not in result.stdout
assert " SUMMARY " in result.stdout
lines = result.stdout.strip().split("\n")
last_line = lines[-1]
assert "== 1 passed in " in last_line
# And the running time is a small positive number
time = float(last_line.split(" ")[-2].replace("s", ""))
assert 0 <= time < 5
示例20
def create_user_open(
*,
username: str = Body(...),
password: str = Body(...),
email: EmailStr = Body(None),
full_name: str = Body(None),
):
"""
Create new user without the need to be logged in.
"""
if not config.USERS_OPEN_REGISTRATION:
raise HTTPException(
status_code=403,
detail="Open user resistration is forbidden on this server",
)
bucket = get_default_bucket()
user = crud.user.get(bucket, username=username)
if user:
raise HTTPException(
status_code=400,
detail="The user with this username already exists in the system",
)
user_in = UserCreate(
username=username, password=password, email=email, full_name=full_name
)
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
示例21
def read_user(username: str, current_user: UserInDB = Depends(get_current_active_user)):
"""
Get a specific user by username (email).
"""
bucket = get_default_bucket()
user = crud.user.get(bucket, username=username)
if user == current_user:
return user
if not crud.user.is_superuser(current_user):
raise HTTPException(
status_code=400, detail="The user doesn't have enough privileges"
)
return user
示例22
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
示例23
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
示例24
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
示例25
def create_item(item_id: str):
if item_id not in items:
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "Some custom header"},
)
return {"item": items[item_id]}
示例26
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return crud.create_user(db=db, user=user)
示例27
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
示例28
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return crud.create_user(db=db, user=user)
示例29
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
示例30
def read_item(item_id: int):
if item_id == 3:
raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
return {"item_id": item_id}
