Adding Datadog to worker and scheduler

compose/fastapi/start-scheduler

@@ -10,5 +10,4 @@ set -o errexit
 set -o pipefail
 set -o nounset
 
-
 taskiq scheduler scheduler:scheduler -fsd

src/broker.py

@@ -19,13 +19,11 @@
 
 class StructlogMiddleware(TaskiqMiddleware):
     def pre_execute(
-            self,
-            message: TaskiqMessage,
+        self,
+        message: TaskiqMessage,
     ) -> "Union[TaskiqMessage, Coroutine[Any, Any, TaskiqMessage]]":
         structlog.contextvars.clear_contextvars()
-        structlog.contextvars.bind_contextvars(
-            taskiq_task_id=message.task_id, taskiq_task_name=message.task_name
-        )
+        structlog.contextvars.bind_contextvars(taskiq_task_id=message.task_id, taskiq_task_name=message.task_name)
 
         return message
 

src/infrastructure/dependency/structured_logs.py

@@ -3,6 +3,7 @@
 
 Much of this is borrowed from: https://gist.github.com/Brymes/cd8f9f138e12845417a246822f64ca26
 """
+
 import logging
 import time
 
@@ -79,10 +80,10 @@ def setup_structured_logging(json_logs: bool = False, log_level: str = "INFO"):
 
     structlog.configure(
         processors=shared_processors
-                   + [
-                       # Prepare event dict for `ProcessorFormatter`.
-                       structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
-                   ],
+        + [
+            # Prepare event dict for `ProcessorFormatter`.
+            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
+        ],
         logger_factory=structlog.stdlib.LoggerFactory(),
         cache_logger_on_first_use=True,
     )
@@ -176,18 +177,19 @@ async def dispatch(self, request: Request, call_next) -> Response:
 
         # Recreate the Uvicorn access log format, but add all parameters as structured information
         logger_fn = access_logger.warn if 400 < status_code < 500 else access_logger.info
-        logger_fn(f"""{real_host}:{client_port} - "{http_method} {url} HTTP/{http_version}" {status_code}""",
-                  http={
-                      "url": str(request.url),
-                      "request_path": str(path),
-                      "status_code": status_code,
-                      "method": http_method,
-                      "request_id": request_id,
-                      "version": http_version,
-                  },
-                  network={"client": {"ip": real_host, "port": client_port}},
-                  duration=process_time,
-                  )
+        logger_fn(
+            f"""{real_host}:{client_port} - "{http_method} {url} HTTP/{http_version}" {status_code}""",
+            http={
+                "url": str(request.url),
+                "request_path": str(path),
+                "status_code": status_code,
+                "method": http_method,
+                "request_id": request_id,
+                "version": http_version,
+            },
+            network={"client": {"ip": real_host, "port": client_port}},
+            duration=process_time,
+        )
 
         # if 400 < status_code < 500:
         #     access_logger.warn(

src/infrastructure/logger.py

@@ -3,7 +3,6 @@
 
 import structlog
 
-
 if os.environ.get("ENV") == "testing":
     # Some tests check logging output, so use the old logger
     fmt = "%(levelname)s:     %(message)s"
@@ -17,5 +16,3 @@
 else:
     # Default to structured logger, enable JSON format if env set
     logger = structlog.stdlib.get_logger("api")  # type: ignore
-
-

src/scheduler.py

@@ -1,6 +1,7 @@
 """
 Entrypoint for the backend scheduler
 """
+
 import datadog  # noqa: F401, I001
 
 from taskiq import TaskiqScheduler

src/worker.py

@@ -1,6 +1,7 @@
 """
 Entrypoint for the backend worker
 """
+
 import datadog  # noqa: F401, I001
 
-from broker import broker as worker # noqa: F401
+from broker import broker as worker  # noqa: F401