Added the first app sketch.

master
blallo 2019-07-30 22:52:44 +02:00 committed by blallo
parent a72f657b56
commit 5e906d53cf
Signed by: blallo
GPG Key ID: 0CBE577C9B72DC3F
3 changed files with 85 additions and 1 deletions

44
api/app.py 100644
View File

@ -0,0 +1,44 @@
# -*- encoding: utf-8 -*-
"""
The application entrypoint.
"""
from aiohttp import web
import click
import logging
import os
import typing as T
from bot_z.async_operator import AsyncOperator
from api.rest import routes
alog = logging.getLogger("api")
alog.setLevel(os.environ.get("BOTZ_LOGLEVEL", logging.INFO))
h = logging.StreamHandler()
f = logging.Formatter("%(levelname)s [%(name)s] -> %(message)s")
h.setFormatter(f)
alog.addHandler(h)
BASE_URI = os.environ.get("BOTZ_BASE_URI")
def run(address: T.Text, port: int) -> None:
"""Application entrypoint."""
app = web.Application(logger=alog)
app["async_operator"] = AsyncOperator(BASE_URI, "test")
app.add_routes(routes)
web.run_app(app, host=address, port=port)
@click.command()
@click.option(
"-a",
"--address",
type=click.STRING,
help="Address to bind the server to.",
default="127.0.0.1",
)
@click.option("-p", "--port", type=click.INT, help="Port to bind to", default=3003)
def cli(address: T.Text, port: int) -> None:
run(address, port)

37
api/rest.py 100644
View File

@ -0,0 +1,37 @@
# -*- encoding: utf-8 -*-
"""
The REST endpoints.
"""
from aiohttp import web
import logging
from api.async_bot import login, logout, checkin, checkout
alog = logging.getLogger("api")
routes = web.RouteTableDef()
@routes.post("/login")
async def login_handler(request: web.Request) -> web.Response:
data = await request.post()
user = data.get("username")
password = data.get("password")
if not user or not password:
return web.json_response({"error": "Missing username or password"}, status=404)
op = request.app["async_operator"]
alog.debug("login - user: %s, password: %s", user, password)
res = await login(op, user, password)
alog.debug("login result: %s", res)
return web.json_response({"logged_in": res}, status=200)
@routes.post("/logout")
async def logout_handler(request: web.Request) -> web.Response:
alog.debug("logout")
op = request.app["async_operator"]
res = await logout(op)
alog.debug("logout result: %s", res)
return web.json_response({"logged_in": res}, status=200)

View File

@ -35,6 +35,9 @@ requirements = [
"selenium>=3.141.0",
"lockfile>=0.12.2",
"python-daemon>=2.2.3",
"aiohttp>=3.5.4",
"aiodns>=2.0.0",
"cchardet",
]
setup_requirements = [] # type: T.List[str]
@ -258,7 +261,7 @@ setup(
"install": CustomInstallCommand,
"bdist_wheel": CustomBDistWheel,
},
entry_points={"console_scripts": ["bot_z=bot_z.cli:cli"]},
entry_points={"console_scripts": ["bot_z=bot_z.cli:cli", "z_app=api.app:cli"]},
package_data={"bot_z": ["bin/geckodriver"]},
include_package_data=True,
install_requires=requirements,