Python aiohttp.web 模块,HTTPInternalServerError() 实例源码
我们从Python开源项目中,提取了以下6个代码示例,用于说明如何使用aiohttp.web.HTTPInternalServerError()。
def __init__(self, app):
self.app = app
env = get_env(app)
if env is None:
text = ("Template engine is not initialized,"
"call aiohttp_jinja2.setup(...) first"
"")
# in order to see meaningful exception message both: on console
# output and rendered page we add same message to *reason* and
# *text* arguments.
raise web.HTTPInternalServerError(reason=text, text=text)
self._set_asset_paths(app)
# We only want to refresh the webpack stats in development mode,
# not everyone sets this setting,so let's assume it's production.
if app.settings.DEBUG:
app.on_response_prepare.append(self._refresh_webpack_stats)
env.globals['asset_url_for'] = self.asset_url_for
def handle_wsdl_request(self, req):
ctx = AioMethodContext(self, req, 'text/xml; charset=utf-8',
aiohttp_app=self._aiohttp_app)
if self.doc.wsdl11 is None:
raise web.HTTPNotFound(headers=ctx.transport.resp_headers)
if self._wsdl is None:
self._wsdl = self.doc.wsdl11.get_interface_document()
ctx.transport.wsdl = self._wsdl
if ctx.transport.wsdl is None:
with self._mtx_build_interface_document:
try:
ctx.transport.wsdl = self._wsdl
if ctx.transport.wsdl is None:
actual_url = urlunparse([req.scheme, req.host, req.path, '', ''])
self.doc.wsdl11.build_interface_document(actual_url)
ctx.transport.wsdl = self._wsdl = self.doc.wsdl11.get_interface_document()
except Exception as e:
logger.exception(e)
ctx.transport.wsdl_error = e
self.event_manager.fire_event('wsdl_exception', ctx)
raise web.HTTPInternalServerError(headers=ctx.transport.resp_headers)
self.event_manager.fire_event('wsdl', ctx)
ctx.transport.resp_headers['Content-Length'] = str(len(ctx.transport.wsdl))
ctx.close()
return await self.make_streaming_response(
req=req,
code=200,
headers=ctx.transport.resp_headers,
content=[ctx.transport.wsdl])
def serialize(self, data):
try:
template.get_env(self.controller.request.app).globals['controller'] = self.controller
return template.render(self.controller._private.template, self.controller.request, data)
except web.HTTPInternalServerError as e:
if self.controller.request.is_ajax():
raise web.HTTPNotImplemented()
else:
raise e
def delete_pet(request):
if request.query.get('petId') == '42':
return web.HTTPInternalServerError()
return web.json_response({})
def get_aiohttp_response(self, context: Dict, default_charset: Optional[str]=None, default_content_type: Optional[str]=None) -> web.Response:
if self.missing_content_type:
self.charset = default_charset
self.content_type = default_content_type
charset = self.charset
if hdrs.CONTENT_TYPE in self._headers and ';' in self._headers[hdrs.CONTENT_TYPE]:
try:
charset = str([v for v in self._headers[hdrs.CONTENT_TYPE].split(';') if 'charset=' in v][0]).replace('charset=', '').strip()
except IndexError:
pass
elif hdrs.CONTENT_TYPE in self._headers and ';' not in self._headers[hdrs.CONTENT_TYPE]:
charset = None
if self._body and not isinstance(self._body, bytes) and charset:
body = self._body
try:
body_value = body.encode(charset.lower())
except (ValueError, LookupError, UnicodeEncodeError) as e:
if not context.get('log_level') or context.get('log_level') in ['DEBUG']:
traceback.print_exception(e.__class__, e, e.__traceback__)
raise web.HTTPInternalServerError() from e
elif self._body:
body_value = self._body.encode() if not isinstance(self._body, bytes) else self._body
else:
body_value = b''
return web.Response(body=body_value, status=self._status, reason=self._reason, headers=self._headers, content_type=self.content_type, charset=self.charset)
def _dispatch(self, action, actionName):
self._private.layout = getattr(self.__class__,
'LAYOUT') if not self.request.is_ajax() else Controller.EMPTY_LAYOUT
self._private.template = os.path.join(self._private.search_path, self._private.controller,
'%s.html' % actionName)
self._private.session = await get_session(self.request)
self._private.flash = Flash(self._private.session)
# Todo: something better
beforeActionRes = {}
for beforeAction in getattr(self.__class__, '__BEFORE_ACTIONS', []):
if actionName not in beforeAction[CtlDecoratorDescriptor.EXCEPT] and \
(actionName in beforeAction[CtlDecoratorDescriptor.ONLY] or
len(beforeAction[CtlDecoratorDescriptor.ONLY]) == 0):
res = await awaitable(beforeAction[CtlDecoratorDescriptor.VAL](self))
if isinstance(res, web.Response):
return res
if isinstance(res, dict):
beforeActionRes.update(res)
try:
res = await awaitable(action())
except web.HTTPException as e:
self._private.flash.sync()
raise e
except Exception as e:
self._private.flash.sync()
raise web.HTTPInternalServerError()
if isinstance(res, web.Response):
self._private.flash.sync()
return res
if res is None:
res = {}
res.update(beforeActionRes)
response = await self.request.serializer.serialize(res)
try:
headers = getattr(self.__class__, '__HEADERS')
for name in headers:
descriptior = headers[name]
if actionName not in descriptior[CtlDecoratorDescriptor.EXCEPT] and \
(actionName in descriptior[CtlDecoratorDescriptor.ONLY] or
len(descriptior[CtlDecoratorDescriptor.ONLY]) == 0):
response.headers[name] = descriptior[CtlDecoratorDescriptor.VAL]
except AttributeError:
pass
self._private.flash.sync()
return response