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 __call__(self, scope, receive, send):
response = PlainTextResponse("OK")
await response(scope, receive, send)
def foo(request):
return PlainTextResponse("Foo")
async def test_get_from_cache_head(cache: Cache) -> None:
scope: Scope = {
"type": "http",
"method": "HEAD",
"path": "/path",
"headers": [],
}
request = Request(scope)
response = PlainTextResponse("Hello, world!")
await store_in_cache(response, request=request, cache=cache)
cached_response = await get_from_cache(request, cache=cache)
assert cached_response is not None
assert ComparableStarletteResponse(cached_response) == response
async def hi(request):
capture_message("hi", level="error")
return PlainTextResponse("ok")
async def plaintext(request, call_next):
if request.url.path == "/":
return PlainTextResponse("OK")
response = await call_next(request)
response.headers["Custom"] = "Example"
return response
# headers in the response.
if self.allow_all_headers and requested_headers is not None:
headers["Access-Control-Allow-Headers"] = requested_headers
elif requested_headers is not None:
for header in [h.lower() for h in requested_headers.split(",")]:
if header.strip() not in self.allow_headers:
failures.append("headers")
# We don't strictly need to use 400 responses here, since its up to
# the browser to enforce the CORS policy, but its more informative
# if we do.
if failures:
failure_text = "Disallowed CORS " + ", ".join(failures)
return PlainTextResponse(failure_text, status_code=400, headers=headers)
return PlainTextResponse("OK", status_code=200, headers=headers)
async def daily_python_list(req):
"""Python 日报列表, 其实就是个按照日期伪造的页面, 用来订阅 rss"""
language = req.path_params['language'].lower()
if language not in {'cn', 'en', 'any'}:
return PlainTextResponse('language should be cn / en / any.')
limit: int = int(req.query_params.get('limit') or 10)
xml_data: dict = {
'channel': {
'title': 'Python Daily',
'description': 'Python Daily Newspaper',
'link': f'https://{ONLINE_HOST}/newspaper/daily.python.list.rss.{language}',
'language': {
'cn': 'zh-cn',
'any': 'zh-cn'
}.get(language, 'en'),
},
'items': []
}
for date_delta in range(1, limit):
title_date: str = ttime(time.time() - 86400 * date_delta)[:10]
# 当日 0 点发布前一天的结果
async def user(request):
user_id = request.path_params["user_id"]
return PlainTextResponse(user_id)
async def method_not_allowed(self, request: Request) -> Response:
# If we're running inside a starlette application then raise an
# exception, so that the configurable exception handler can deal with
# returning the response. For plain ASGI apps, just return the response.
if "app" in self.scope:
raise HTTPException(status_code=405)
return PlainTextResponse("Method Not Allowed", status_code=405)