BotZ/api/app.py

103 lines
2.9 KiB
Python
Raw Normal View History

2019-07-30 22:52:44 +02:00
# -*- encoding: utf-8 -*-
"""
The application entrypoint.
"""
from aiohttp import web
2019-08-01 00:29:28 +02:00
import base64
from cryptography import fernet
2019-07-30 22:52:44 +02:00
import logging
2019-08-06 10:04:02 +02:00
import logging.handlers
2019-07-30 22:52:44 +02:00
import os
import typing as T
2019-08-01 00:29:28 +02:00
import click
from aiohttp_session import setup, session_middleware
from aiohttp_session.cookie_storage import EncryptedCookieStorage
2019-07-30 22:52:44 +02:00
from bot_z.async_operator import AsyncOperator
from api.rest import routes, add_static_routes
2019-08-06 10:04:02 +02:00
from api.conf import read_conf
2019-07-30 22:52:44 +02:00
2019-08-06 10:04:02 +02:00
def setup_log(level: str, syslog: bool) -> logging.Logger:
2019-08-01 00:29:28 +02:00
alog = logging.getLogger("api")
2019-08-06 10:04:02 +02:00
alog.setLevel(os.environ.get("BOTZ_LOGLEVEL", level))
2019-08-01 00:29:28 +02:00
h = logging.StreamHandler()
f = logging.Formatter("%(levelname)s [%(name)s] -> %(message)s")
h.setFormatter(f)
alog.addHandler(h)
2019-08-06 10:04:02 +02:00
if syslog:
sh = logging.handlers.SysLogHandler()
alog.addHandler(sh)
2019-08-01 00:29:28 +02:00
return alog
def init_secret() -> bytes:
fernet_key = fernet.Fernet.generate_key()
return base64.urlsafe_b64decode(fernet_key)
2019-08-06 10:04:02 +02:00
def setup_session(app: web.Application, secure: bool):
2019-08-01 00:29:28 +02:00
secret = init_secret()
2019-08-06 10:04:02 +02:00
setup(
app,
EncryptedCookieStorage(
secret_key=secret, cookie_name="BOTZ_SESSION", httponly=False, secure=secure
),
)
2019-07-30 22:52:44 +02:00
2019-08-06 10:04:02 +02:00
def run(
address: T.Optional[T.Text], port: T.Optional[int], conf_path: T.Optional[T.Text]
) -> None:
2019-07-30 22:52:44 +02:00
"""Application entrypoint."""
2019-08-06 10:04:02 +02:00
conf = read_conf(conf_path)
# This closure is needed to intercept the to-be-prepared response
# and add the right CORS headers
async def on_prepare_cors(request, response):
response.headers["Access-Control-Allow-Origin"] = conf["http"].get("cors_allow")
response.headers["Access-Control-Allow-Credentials"] = "true"
alog = setup_log(conf["log"]["level"], conf["log"]["syslog"])
alog.debug("conf %s", conf)
2019-07-30 22:52:44 +02:00
app = web.Application(logger=alog)
2019-08-06 10:04:02 +02:00
app["base_uri"] = conf["base_uri"]
app["debug"] = conf["debug"]
app["headless"] = conf["headless"]
if conf["http"].get("cors_allow"):
app.on_response_prepare.append(on_prepare_cors)
setup_session(app, conf["http"]["cookie_secure"])
add_static_routes(alog)
2019-07-30 22:52:44 +02:00
app.add_routes(routes)
2019-08-06 10:04:02 +02:00
addr = []
if address is not None:
addr.append(address)
if conf["http"].get("bind_addr"):
addr.extend(conf["http"]["bind_addr"])
if port is None:
port = conf["http"]["port"]
alog.debug("Starting app with: addr -> %s, port -> %d", addr, port)
web.run_app(app, host=addr, port=port)
2019-07-30 22:52:44 +02:00
@click.command()
@click.option(
2019-08-06 10:04:02 +02:00
"-a", "--address", type=click.STRING, help="Address to bind the server to."
)
@click.option("-p", "--port", type=click.INT, help="Port to bind to.")
@click.option(
"-c",
"--conf",
type=click.Path(exists=False),
help="A path to a configuration file.",
2019-07-30 22:52:44 +02:00
)
2019-08-06 10:04:02 +02:00
def cli(
address: T.Optional[T.Text],
port: T.Optional[int] = None,
conf: T.Optional[T.Text] = None,
) -> None:
run(address, port, conf)