Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_repr():
assert repr(HTTPException(404)) == (
"HTTPException(status_code=404, detail='Not Found')"
)
assert repr(HTTPException(404, detail="Not Found: foo")) == (
"HTTPException(status_code=404, detail='Not Found: foo')"
)
class CustomHTTPException(HTTPException):
pass
assert repr(CustomHTTPException(500, detail="Something custom")) == (
"CustomHTTPException(status_code=500, detail='Something custom')"
)
def check_can_edit(request, username):
can_edit = request.session.get("username") == username
if request.method in ("POST", "PUT", "PATCH", "DELETE") and not can_edit:
raise HTTPException(status_code=403)
return can_edit
description: The requested edge
400:
description: The requested edge doesn't exit
"""
state = request.app.state
graph = state.graph
node_deserializer = state.settings.node_deserializer
path_params = request.path_params
u, v = path_params['u'], path_params['v']
if node_deserializer is not None:
u, v = node_deserializer(u), node_deserializer(v)
try:
data = graph.get_edge(u, v)
except KeyError:
raise HTTPException(404, f'Edge ({u}, {v}) not found in graph')
return JSONResponse(data)
async def async_wrapper(
*args: typing.Any, **kwargs: typing.Any
) -> Response:
request = kwargs.get("request", args[idx])
assert isinstance(request, Request)
if not has_required_scope(request, scopes_list):
if redirect is not None:
return RedirectResponse(url=request.url_for(redirect))
raise HTTPException(status_code=status_code)
return await func(*args, **kwargs)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http" or scope["path"] not in ("/", "/get"):
raise HTTPException(404)
request = Request(scope, receive, send)
if scope["path"] == "/get":
handler = getattr(self, "docs")
elif scope["path"] == "/":
handler = getattr(self, "template")
response = await handler(request)
await response(scope, receive, send)
async def app(request: Request) -> Response:
try:
body = None
if body_field:
if is_body_form:
body = await request.form()
else:
body_bytes = await request.body()
if body_bytes:
body = await request.json()
except Exception as e:
logger.error(f"Error getting request body: {e}")
raise HTTPException(
status_code=400, detail="There was an error parsing the body"
) from e
solved_result = await solve_dependencies(
request=request,
dependant=dependant,
body=body,
dependency_overrides_provider=dependency_overrides_provider,
)
values, errors, background_tasks, sub_response, _ = solved_result
if errors:
raise RequestValidationError(errors)
else:
assert dependant.call is not None, "dependant.call must be a function"
if is_coroutine:
raw_response = await dependant.call(**values)
else:
async def wrapper(request: Request, **kwargs) -> Response:
openapi_request = await StarletteOpenAPIRequest(request)
validated_request = RequestValidator(
self.spec, custom_formatters=self.custom_formatters,
custom_media_type_deserializers=self.custom_media_type_deserializers
).validate(openapi_request)
try:
validated_request.raise_for_errors()
except InvalidSecurity as ex:
raise HTTPException(HTTPStatus.FORBIDDEN, "Invalid security.") from ex
except OpenAPIError as ex:
raise HTTPException(HTTPStatus.BAD_REQUEST, "Bad request") from ex
response = endpoint_fn(request, **kwargs)
if iscoroutine(response):
response = await response
if isinstance(response, dict):
response = JSONResponse(response)
elif not isinstance(response, Response):
raise ValueError(
f"The endpoint function `{endpoint_fn.__name__}` must return"
" either a dict or a Starlette Response instance."
)
# TODO: pass a list of operation IDs to specify which responses not to validate
if self.validate_responses:
ResponseValidator(
self.spec, custom_formatters=self.custom_formatters,
title=params.site_title,
description=params.site_description,
version=__version__,
default_response_class=JSONResponse,
**DOCS_PARAMS,
)
# Add Event Handlers
for startup in on_startup:
app.add_event_handler("startup", startup)
for shutdown in on_shutdown:
app.add_event_handler("shutdown", shutdown)
# HTTP Error Handler
app.add_exception_handler(StarletteHTTPException, http_handler)
# Backend Application Error Handler
app.add_exception_handler(HyperglassError, app_handler)
# Request Validation Error Handler
app.add_exception_handler(RequestValidationError, validation_handler)
# App Validation Error Handler
app.add_exception_handler(ValidationError, validation_handler)
# Uncaught Error Handler
app.add_exception_handler(Exception, default_handler)
def _custom_openapi():
"""Generate custom OpenAPI config."""
from starlette.exceptions import HTTPException
class Http404(HTTPException):
def __init__(self, detail: str = None) -> None:
super().__init__(404, detail=detail)
def paginate(cls, request, objects):
paginator = cls.paginator_class(objects, cls.paginate_by)
page_number = request.query_params.get("page")
try:
page_number = int(page_number)
except (TypeError, ValueError):
page_number = 1
try:
page = paginator.page(page_number)
return (paginator, page, page.object_list, page.has_other_pages)
except InvalidPage as e:
raise HTTPException(404, f"Invalid page {page_number}: {str(e)}")