我们从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'], styles_css_url=request.app['static_root_url'] + '/styles.css', 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, 'timestamp': ts, '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', '/api/v{}/configure'.format(VERSION), 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', '/api/v1/action/{action_name:preflight|postflight|deploy}', 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', styles_css_url=request.app['static_root_url'] + '/styles.css', 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 %}
