Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_named_resources(swagger_docs):
async def handler(request):
return web.json_response()
swagger = swagger_docs()
swagger.add_routes(
[web.get("/", handler, name="get"), web.post("/", handler, name="post")]
)
assert "get" in swagger._app.router
assert "post" in swagger._app.router
@asyncio.coroutine
def init(loop, address, port):
app = web.Application(loop=loop)
app.router.add_route('GET', '/', handle)
server = yield from loop.create_server(app.make_handler(),
address, port)
host = server.sockets[0].getsockname()
print('Serving on {}. Hit CTRL-C to stop.'.format(host))
"""Serve plugins for lovelace."""
try:
# Strip '?' from URL
if "?" in requested_file:
requested_file = requested_file.split("?")[0]
file = "{}/www/community/{}".format(self.config_dir, requested_file)
# Serve .gz if it exist
if os.path.exists(file + ".gz"):
file += ".gz"
response = None
if os.path.exists(file):
_LOGGER.debug("Serving %s from %s", requested_file, file)
response = web.FileResponse(file)
response.headers["Cache-Control"] = "max-age=0, must-revalidate"
else:
_LOGGER.debug("Tried to serve up '%s' but it does not exist", file)
raise HTTPNotFound()
except Exception as error: # pylint: disable=broad-except
_LOGGER.debug(
"there was an issue trying to serve %s - %s", requested_file, error
)
raise HTTPNotFound()
return response
asset = assets.get(path, None)
# If there is no such asset, return 404 not found
if asset is None:
return web.Response(status=404, text='not found')
# Otherwise, get the content type and return
ct = mimetypes.guess_type(path)[0] or 'application/octet-stream'
return web.Response(body=asset, content_type=ct)
if __name__ == '__main__':
# Here are some aiohttp specifics. Note that all assets except the
# main app are prefixed with "flexx/...", we can use that in the routing.
app = web.Application()
app.router.add_get('/', handler)
app.router.add_get('/{tail:flexx/.*}', handler)
web.run_app(app, host='0.0.0.0', port=8080)
def wrapper(request):
#import pudb; pudb.set_trace()
arguments = dict(request.match_info)
try:
headers = {
key.lower(): value
for key, value in request.headers.items()}
response = handler(headers, **arguments)
if isinstance(response, dict):
return aiohttp.web.Response(
content_type='application/json',
status=200,
text=json.dumps(response))
else:
return aiohttp.web.Response(
body=response.body,
status=response.status,
headers=response.headers)
except Exception as e:
log.exception('An error occurred when handling request')
raise aiohttp.web.HTTPInternalServerError()
env = Env()
resp = yield from aiohttp.request(
'GET', '%s/info/' % env('host_web'),
headers=request.headers,
cookies=request.cookies.items(),
allow_redirects=False
)
if resp.status == 200:
data = yield from resp.json()
username = data.get('username')
if username:
env.username = username
return env, None
else:
return None, web.Response(body=b'403 Forbidden', status=403)
body = yield from resp.read()
return None, web.Response(body=body, status=resp.status)
def error(status, msg, *args):
"""
Logs an error message and returns an error to
the web client.
Args:
status (int): The http response status.
msg (str): The message to display.
args (tuple): Arguments for string format on msg.
"""
logger.error(msg.format(*args))
return web.Response(status=status, text=msg.format(*args))
def main():
# Make sure Lightbus formats its logs correctly
lightbus.configure_logging()
# Create our lightbus client and our web application
bus = lightbus.create()
app = web.Application()
app.router.add_route("GET", "/", home_view)
app.on_startup.append(start_listener)
app.on_cleanup.append(cleanup)
# Store the bus on `app` as we'll need it
# in start_listener() and cleanup()
app.bus = bus
web.run_app(app, host="127.0.0.1", port=5000)
code such as that for restarting the server after software update. This custom runner allows handling of signals
as well as restart and shutdown events from users.
https://docs.aiohttp.org/en/stable/web_advanced.html#application-runners
:param app: the application
:param host: the host to listen on
:param port: the port to listen on
:return: a custom :class:`~aiohttp.web.AppRunner`
"""
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, host, port)
await site.start()
logger.info(f"Listening at http://{host}:{port}")
return runner
def run():
sass_compile('notebook')
root = os.path.dirname(os.path.abspath(__file__))
# notebook
notebook_app = web.Application()
notebook_app.on_startup.append(on_startup)
setup_aiohttp_jinja2(notebook_app, 'notebook')
setup_aiohttp_session(notebook_app)
routes.static('/static', f'{root}/static')
setup_common_static_routes(routes)
notebook_app.add_routes(routes)
# workshop
workshop_app = web.Application()
workshop_app.on_startup.append(on_startup)
setup_aiohttp_jinja2(workshop_app, 'notebook')