# -*- encoding: utf-8 -*- from aiohttp import web import logging import click logging.basicConfig( format="%(message)s", level=logging.INFO, handlers=[logging.StreamHandler()] ) alog = logging.getLogger(__name__) routes = web.RouteTableDef() @routes.get("/{tail:.*}") async def get_handler(request: web.Request) -> web.Response: data = await request.get() alog.info("GET -> [%s]: %s", request.path, data) return web.json_response({"method": request.method, "path": request.path}) @routes.post("/{tail:.*}") async def post_handler(request: web.Request) -> web.Response: data = await request.post() alog.info("POST -> [%s]: %s", request.path, data) return web.json_response({"method": request.method, "path": request.path}) @routes.put("/{tail:.*}") async def put_handler(request: web.Request) -> web.Response: data = await request.put() alog.info("PUT -> [%s]: %s", request.path, data) return web.json_response({"method": request.method, "path": request.path}) @routes.delete("/{tail:.*}") async def delete_handler(request: web.Request) -> web.Response: alog.info("DELETE -> [%s]", request.path) return web.json_response({"method": request.method, "path": request.path}) async def options_handler(request: web.Request) -> web.Response: alog.info("OPTIONS -> [%s]", request.path) return web.Response(status=200, headers={"Access-Control-Allow-Origin": "*"}) def run(address: str, port: int) -> None: """Application entrypoint.""" app = web.Application(logger=alog) app.add_routes(routes) app.router.add_route("OPTIONS", "/{tail:.*}", options_handler) 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: str, port: int) -> None: run(address, port) if __name__ == "__main__": cli()