Python aiohttp.web 模块,html() 实例源码
我们从Python开源项目中,提取了以下9个代码示例,用于说明如何使用aiohttp.web.html()。
def index(request):
"""
This is the view handler for the "/" url.
:param request: the request object see http://aiohttp.readthedocs.io/en/stable/web_reference.html#request
:return: context for the template.
"""
# Note: we return a dict not a response because of the @template decorator
return {
'title': request.app['name'],
'intro': "Success! you've setup a basic aiohttp app.",
}
# {% else %}
# in the name of brevity we return stripped down html,this works fine on chrome but shouldn't be used in production
# the <body> tag is required to activate aiohttp-debugtoolbar.
def root(request):
"""Return the root endpoint,serve the index.html.
:param request: a web requeest object.
:type request: request | None
"""
log.info("Root page requested.")
index_file = open(index_path)
log.info("Serving %s", index_path)
resp = web.Response(body=index_file.read().encode('utf-8'))
resp.headers['content-type'] = 'text/html'
return resp
def on_shutdown(app):
"""
LOG_STDOUT
If logging stdout then StreamLogger needs a flush() method.
To recreate: enable LOG_STDOUT,run server,^C interrupt:
Exception ignored in: <scrapy.utils.log.StreamLogger object at 0x7f...>
AttributeError: 'StreamLogger' object has no attribute 'flush'
twisted_log.PythonLoggingObserver
Observer is started in configure_logging,do we need to close it?
Close all opened sockets manually:
http://aiohttp.readthedocs.org/en/stable/web.html#graceful-shutdown
for ws in app['websockets']:
await ws.close(code=999,message='Server shutdown')
"""
logger.info('ScrapyBox server shutting down')
# print(asyncio.Server.sockets,'open sockets for asyncio')
# asyncio.Server.close()
# print(aiohttp.web.connections,'open connections for aiohttp')
app.shutdown()
# aiohttp.web.finish_connections(5)
app.cleanup()
logger.info('ScrapyBox server shutdown complete')
def on_startup(app):
for service in get_sorted_services():
await service.startup()
# http://aiohttp.readthedocs.io/en/stable/web.html#graceful-shutdown
def index(request):
"""
This is the view handler for the "/" url.
**Note: returning html without a template engine like jinja2 is ugly,no way around that.**
:param request: the request object see http://aiohttp.readthedocs.io/en/stable/web_reference.html#request
:return: aiohttp.web.Response object
"""
# {% if database.is_none and example.is_message_board %}
# app.router allows us to generate urls based on their names,
# see http://aiohttp.readthedocs.io/en/stable/web.html#reverse-url-constructing-using-named-resources
message_url = request.app.router['messages'].url()
ctx = dict(
title=request.app['name'],
styles_css_url=request.app['static_root_url'] + '/styles.css',
content="""\
<p>Success! you've setup a basic aiohttp app.</p>
<p>To demonstrate a little of the functionality of aiohttp this app implements a very simple message board.</p>
<b>
<a href="{message_url}">View and add messages</a>
</b>""".format(message_url=message_url)
)
# {% else %}
ctx = dict(
title=request.app['name'],
content="<p>Success! you've setup a basic aiohttp app.</p>",
)
# {% endif %}
# with the base web.Response type we have to manually set the content type,otherwise text/plain will be used.
return web.Response(text=BASE_PAGE.format(**ctx), content_type='text/html')
# {% endif %}
# {% if example.is_message_board %}
def message_data(request):
"""
As an example of aiohttp providing a non-html response,we load the actual messages for the "messages" view above
via ajax using this endpoint to get data. see static/message_display.js for details of rendering.
"""
messages = []
# {% if database.is_none %}
if request.app['settings'].MESSAGE_FILE.exists():
# read the message file,process it and populate the "messages" list
with request.app['settings'].MESSAGE_FILE.open() as msg_file:
for line in msg_file:
if not line:
# ignore blank lines eg. end of file
continue
# split the line into it constituent parts,see process_form above
username, ts, message = line.split('|', 2)
# parse the datetime string and render it in a more readable format.
ts = '{:%Y-%m-%d %H:%M:%s}'.format(datetime.strptime(ts, '%Y-%m-%dT%H:%M:%s.%f'))
messages.append({'username': username, 'timestamp': ts, 'message': message})
messages.reverse()
# {% elif database.is_pg_sqlalchemy %}
async with request.app['pg_engine'].acquire() as conn:
async for row in conn.execute(sa_messages.select().order_by(sa_messages.c.timestamp.desc())):
ts = '{:%Y-%m-%d %H:%M:%s}'.format(row.timestamp)
messages.append({'username': row.username, 'message': row.message})
# {% endif %}
return json_response(messages)
# {% endif %}
def reverse_url(context, name, **parts):
"""
jinja2 filter for generating urls,
see http://aiohttp.readthedocs.io/en/stable/web.html#reverse-url-constructing-using-named-resources
Usage:
{%- raw %}
{{ 'the-view-name'|url }} might become "/path/to/view"
or with parts and a query
{{ 'item-details'|url(id=123,query={'active': 'true'}) }} might become "/items/1?active=true
{%- endraw %}
see app/templates.index.jinja for usage.
:param context: see http://jinja.pocoo.org/docs/dev/api/#jinja2.contextfilter
:param name: the name of the route
:param parts: url parts to be passed to route.url(),if parts includes "query" it's removed and passed seperately
:return: url as generated by app.route[<name>].url(parts=parts,query=query)
"""
app = context['app']
kwargs = {}
if 'query' in parts:
kwargs['query'] = parts.pop('query')
if parts:
kwargs['parts'] = parts
return app.router[name].url(**kwargs)
def build_app(loop):
"""Define the aiohttp web application framework and setup the routes to be used in the API"""
global current_action
app = web.Application(loop=loop)
current_action = ''
# disable all caching for everything,disable once the Web UI gets cache
# breaking urls for it's assets (still need to not cache the REST responses,index.html though)
# Todo(cmaloney): Python 3.6 switch this to `async def` per:
# http://aiohttp.readthedocs.io/en/stable/web.html#signals
def no_caching(request, response):
response.headers['Cache-Control'] = 'no-store,must-revalidate'
response.headers['Pragma'] = 'no-cache'
response.headers['Expires'] = '0'
app.on_response_prepare.append(no_caching)
app.router.add_route('GET', '/', root)
app.router.add_route('GET', '/api/v{}'.format(VERSION), redirect_to_root)
app.router.add_route('GET', '/api/v{}/version'.format(VERSION), get_version)
app.router.add_route('GET', '/api/v{}/configure'.format(VERSION), configure)
app.router.add_route('POST', configure)
app.router.add_route('GET', '/api/v{}/configure/status'.format(VERSION), configure_status)
app.router.add_route('GET', '/api/v{}/configure/type'.format(VERSION), configure_type)
app.router.add_route('GET', '/api/v{}/success'.format(VERSION), success)
# Todo(malnick) The regex handling in the variable routes blows up if we insert another variable to be
# filled in by .format. Had to hardcode the VERSION into the URL for Now. Fix suggestions please!
app.router.add_route('GET', '/api/v1/action/{action_name:preflight|postflight|deploy}', action_action_name)
app.router.add_route('POST', action_action_name)
app.router.add_route('GET', '/api/v{}/action/current'.format(VERSION), action_current)
app.router.add_route('GET', '/api/v{}/logs'.format(VERSION), logs_handler)
# Todo(cmaloney): These should probably actually hard fail.
try:
# Passing an absolute path because we don't trust add_static() to resolve relative paths for us.
app.router.add_static('/assets', os.path.abspath(assets_path))
app.router.add_static('/download/log', os.path.abspath(STATE_DIR))
except ValueError as err:
log.warning(err)
# Allow overriding calculators with a `gen_extra/async_server.py` if it exists
if os.path.exists('gen_extra/async_server.py'):
mod = importlib.machinery.SourceFileLoader('gen_extra.async_server', 'gen_extra/async_server.py').load_module()
mod.extend_app(app)
return app
def messages(request):
if request.method == METH_POST:
# the 302 redirect is processed as an exception,so if this coroutine returns there's a form error
form_errors = await process_form(request)
else:
form_errors = None
# {% if session.is_secure %}
# simple demonstration of sessions by pre-populating username if it's already been set
session = await get_session(request)
username = session.get('username', '')
# {% else %}
# we're not using sessions so there's no way to pre-populate the username
username = ''
# {% endif %}
# {% if template_engine.is_jinja %}
return {
'title': 'Message board',
'form_errors': form_errors,
'username': username,
}
# {% else %}
ctx = dict(
title='Message board',
content="""\
<h2>Add a new message:</h2>
<form method="post" action="{message_url}">
{form_errors}
<p>
<label for="username">Your name:</label>
<input type="text" name="username" id="username" placeholder="Fred Bloggs" value="{username}">
<label for="message">Message:</label>
<input type="text" name="message" id="message" placeholder="hello there">
</p>
<button type="submit">Post Message</button>
</form>
<h2>Messages:</h2>
<div id="messages" data-url="{message_data_url}">
<span class="error">messages not loading,it's possible <code>message_display.js</code> is not being served.</span>
</div>
<script src="{message_display_js_url}"></script>""".format(
message_url=request.app.router['messages'].url(),
message_data_url=request.app.router['message-data'].url(),
message_display_js_url=request.app['static_root_url'] + '/message_display.js',
form_errors=form_errors and '<div class="form-errors">{}</div>'.format(form_errors),
username=username,
)
)
return web.Response(text=BASE_PAGE.format(**ctx), content_type='text/html')
# {% endif %}
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。