Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def exception_handler(self, request, http_exception):
nonlocal app
assert self is app
return Response(200, content=TextContent('Called'))
async def example(request):
nonlocal calls
calls.append(5)
return Response(200, [(b'Server', b'Python/3.7')], JsonContent({'id': '123'}))
def get_files_list_html_response(template, parent_folder_path, contents):
info = ''
for item in contents:
full_rel_path = html.escape(os.path.join(parent_folder_path, item.get("rel_path")))
info += f'<li><a href="/{full_rel_path}">{full_rel_path}</a></li>'
p = []
whole_p = []
for fragment in parent_folder_path.split('/'):
if fragment:
whole_p.append(html.escape(fragment))
fragment_path = '/'.join(whole_p)
p.append(f'<a href="/{fragment_path}">{html.escape(fragment)}</a>')
# TODO: use chunked encoding here with HTML response
return Response(200, content=HtmlContent(template.format_map({'path': '/'.join(p), 'info': info})))
def status_code(status: int = 200, message: MessageType = None):
"""Returns a plain response with given status, with optional message; sent as plain text or JSON."""
if not message:
return Response(status)
if isinstance(message, str):
content = TextContent(message)
else:
content = JsonContent(message)
return Response(status, content=content)
def no_content():
"""Returns an HTTP 204 No Content response."""
return Response(204)
if requested_range:
_validate_range(requested_range, info.size)
headers = [
(b'Last-Modified', info.modified_time.encode()),
(b'ETag', current_etag),
(b'Accept-Ranges', b'bytes')
]
if cache_time > 0:
headers.append((b'Cache-Control', b'max-age=' + str(cache_time).encode()))
if previous_etag and current_etag == previous_etag:
# handle HTTP 304 Not Modified
return Response(304, headers, None)
if request.method == 'HEAD':
# NB: responses to HEAD requests don't have a body,
# and responses with a body in BlackSheep have content-type and content-length headers set automatically,
# depending on their content; therefore here it's necessary to set content-type and content-length for HEAD
headers.append((b'Content-Type', info.mime.encode()))
headers.append((b'Content-Length', str(info.size).encode()))
return Response(200, headers, None)
status = 200
mime = get_mime_type(resource_path).encode()
if requested_range and is_requested_range_actual(request, info):
# NB: the method can only be GET for range requests, so it cannot happen
# to have response 206 partial content with HEAD
status = 206
async def handle_authentication_challenge(app,
request,
exception: AuthenticateChallenge):
return Response(401,
[exception.get_header()],
content=TextContent('Unauthorized'))
def status_code(status: int = 200, message: MessageType = None):
"""Returns a plain response with given status, with optional message; sent as plain text or JSON."""
if not message:
return Response(status)
if isinstance(message, str):
content = TextContent(message)
else:
content = JsonContent(message)
return Response(status, content=content)