From 3dce5305e7c441fef09fdaa6900d4b2ce8da6370 Mon Sep 17 00:00:00 2001 From: Justin Ryburn Date: Fri, 9 Feb 2024 09:18:05 -0600 Subject: [PATCH] Lint cleanups --- plugins/event_source/kentik_webhook.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/plugins/event_source/kentik_webhook.py b/plugins/event_source/kentik_webhook.py index f4aa408..3dd8f57 100644 --- a/plugins/event_source/kentik_webhook.py +++ b/plugins/event_source/kentik_webhook.py @@ -22,6 +22,7 @@ logger = logging.getLogger(__name__) routes = web.RouteTableDef() + # Initialize the loggger configuration def _initialize_logger_config() -> None: logging.basicConfig( @@ -30,11 +31,12 @@ def _initialize_logger_config() -> None: datefmt="%Y-%m-%d %I:%M:%S", ) + # Process incoming webhook alert notifications from Kentik @routes.post("/alert") async def webhook(request: web.Request) -> web.Response: """Process incoming webhook alert notifications from Kentik - + Parameters ---------- request : web.Request @@ -48,7 +50,7 @@ async def webhook(request: web.Request) -> web.Response: ------ HTTPBadRequest If the payload can't be parsed as JSON - + """ logger.info("Received alert") try: @@ -66,6 +68,7 @@ async def webhook(request: web.Request) -> web.Response: await request.app["queue"].put(data) return web.json_response({}) + # Set app_attr settings in a dictionary def set_app_attr(args: dict[str, Any]) -> dict[str, Any]: """Set app_attr settings in a dictionary @@ -74,12 +77,12 @@ def set_app_attr(args: dict[str, Any]) -> dict[str, Any]: ---------- args : Dict[str,Any] Empty dictionary of arguments - + Returns ------- args : Dict[str,Any] Args containing the host and port - + """ if "host" not in args: host="0.0.0.0" @@ -91,10 +94,11 @@ def set_app_attr(args: dict[str, Any]) -> dict[str, Any]: return app_attrs + # Entrypoint from ansible-rulebook async def main(queue: asyncio.Queue, args: dict[str, Any]) -> None: """Receive events via webhook. - + Parameters ---------- queue : asyncio.Queue