import logging
import asyncio
import html
from enum import Enum
from aiohttp import web
from aiohttp.web import HTTPFound, HTTPMovedPermanently, HTTPException
from aiohttp.web_exceptions import HTTPNotFound
from ..util import stringify, format_tb, Addr
from ..config import is_development_mode
_logger = logging.getLogger(__name__)
def _render_error(error_type, error_message, traceback=None, code=500):
traceback_html = ''
if traceback:
traceback = '\n\n'.join(traceback)
traceback_html = f"""
Traceback
{html.escape(traceback)}
"""
buf = f"""
Error: {html.escape(error_type)}
{html.escape(error_type)}
{html.escape(error_message)}
{traceback_html}
"""
return web.Response(text=buf, status=code, content_type='text/html')
@web.middleware
async def errors_handler_middleware(request, handler):
try:
response = await handler(request)
return response
except HTTPNotFound:
return _render_error(
error_type='Not Found',
error_message='The page you requested has not been found.',
code=404
)
except (HTTPFound, HTTPMovedPermanently) as exc:
raise exc
except HTTPException as exc:
_logger.exception(exc)
return _render_error(
error_type=exc.reason,
error_message=exc.text,
traceback=format_tb(exc)
)
except Exception as exc:
_logger.exception(exc)
return _render_error(
error_type=exc.__class__.__name__,
error_message=exc.message if hasattr(exc, 'message') else str(exc),
traceback=format_tb(exc)
)
def serve(addr: Addr, before_start=None, handle_signals=True, routes=None, event_loop=None):
logging.getLogger('aiohttp').setLevel(logging.DEBUG if is_development_mode() else logging.WARNING)
app = web.Application()
app.middlewares.append(errors_handler_middleware)
if routes is not None:
app.add_routes(routes)
if callable(before_start):
before_start(app)
if not event_loop:
event_loop = asyncio.get_event_loop()
runner = web.AppRunner(app, handle_signals=handle_signals)
event_loop.run_until_complete(runner.setup())
host, port = addr
site = web.TCPSite(runner, host=host, port=port)
event_loop.run_until_complete(site.start())
_logger.info(f'Server started at http://{host}:{port}')
event_loop.run_forever()
def ajax_ok(data=None):
if data is None:
data = 1
response = {'response': data}
return web.json_response(response, dumps=stringify)
class HTTPMethod(Enum):
GET = 'GET'
POST = 'POST'
PUT = 'PUT'