def static_request_handler(cls: Any, obj: Any, context: Dict, func: Any, path: str, base_url: str) -> Any:
if '?P<filename>' not in base_url:
pattern = r'^{}(?P<filename>.+?)$'.format(re.sub(r'\$$', '', re.sub(r'^\^?(.*)$', r'\1', base_url)))
else:
pattern = r'^{}$'.format(re.sub(r'\$$', '', re.sub(r'^\^?(.*)$', r'\1', base_url)))
compiled_pattern = re.compile(pattern)
if path.startswith('/'):
path = os.path.dirname(path)
else:
path = '{}/{}'.format(os.path.dirname(context.get('context', {}).get('_service_file_path')), path)
if not path.endswith('/'):
path = '{}/'.format(path)
async def handler(request: web.Request) -> web.Response:
result = compiled_pattern.match(request.path)
filename = result.groupdict()['filename']
filepath = '{}{}'.format(path, filename)
try:
if os.path.isdir(filepath) or not os.path.exists(filepath):
raise web.HTTPNotFound()
pathlib.Path(filepath).open('r')
return FileResponse(filepath)
except PermissionError as e:
raise web.HTTPForbidden()
context['_http_routes'] = context.get('_http_routes', [])
context['_http_routes'].append(('GET', pattern, handler))
start_func = cls.start_server(obj, context)
return (await start_func) if start_func else None
评论列表
文章目录