http.py 文件源码

def error_handler(cls: Any, obj: Any, context: Dict, func: Any, status_code: int) -> Any:
        default_content_type = context.get('options', {}).get('http', {}).get('content_type', 'text/plain')
        default_charset = context.get('options', {}).get('http', {}).get('charset', 'utf-8')

        if default_content_type is not None and ";" in default_content_type:
            # for backwards compability
            try:
                default_charset = str([v for v in default_content_type.split(';') if 'charset=' in v][0]).replace('charset=', '').strip()
                default_content_type = str([v for v in default_content_type.split(';')][0]).strip()
            except IndexError:
                pass

        async def handler(request: web.Request) -> web.Response:
            values = inspect.getfullargspec(func)
            kwargs = {k: values.defaults[i] for i, k in enumerate(values.args[len(values.args) - len(values.defaults):])} if values.defaults else {}
            routine = func(*(obj, request,), **kwargs)
            return_value = (await routine) if isinstance(routine, Awaitable) else routine  # type: Union[str, bytes, Dict, List, Tuple, web.Response, Response]

            if isinstance(return_value, Response):
                return return_value.get_aiohttp_response(context, default_content_type=default_content_type, default_charset=default_charset)

            status = int(status_code)
            headers = None

            if isinstance(return_value, dict):
                body = return_value.get('body')
                _status = return_value.get('status')  # type: Optional[SupportsInt]
                if _status and isinstance(_status, (int, str, bytes)):
                    status = int(_status)
                if return_value.get('headers'):
                    headers = CIMultiDict(return_value.get('headers'))
            elif isinstance(return_value, list) or isinstance(return_value, tuple):
                _status = return_value[0]
                if _status and isinstance(_status, (int, str, bytes)):
                    status = int(_status)
                body = return_value[1]
                if len(return_value) > 2:
                    headers = CIMultiDict(return_value[2])
            elif isinstance(return_value, web.Response):
                return return_value
            else:
                if return_value is None:
                    return_value = ''
                body = return_value

            return Response(body=body, status=status, headers=headers, content_type=default_content_type, charset=default_charset).get_aiohttp_response(context)

        context['_http_error_handler'] = context.get('_http_error_handler', {})
        context['_http_error_handler'][int(status_code)] = handler

        start_func = cls.start_server(obj, context)
        return (await start_func) if start_func else None