1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
import logging
import asyncio
import html
from enum import Enum
from aiohttp import web
from aiohttp.web import HTTPFound
from aiohttp.web_exceptions import HTTPNotFound
from ..util import stringify, format_tb, Addr
_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"""
<div class="error_traceback">
<div class="error_title">Traceback</div>
<div class="error_traceback_content">{html.escape(traceback)}</div>
</div>
"""
buf = f"""
<!doctype html>
<html lang=en>
<head>
<title>Error: {html.escape(error_type)}</title>
<meta http-equiv="content-type" content="text/html; charset=utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0, minimum-scale=1.0, maximum-scale=1.0, user-scalable=no">
<link rel="stylesheet" type="text/css" href="/assets/error_page.css">
</head>
<body>
<div class="error_title">{html.escape(error_type)}</div>
<div class="error_message">{html.escape(error_message)}</div>
{traceback_html}
</body>
</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 as exc:
raise 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):
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'
|
