save

bolt-jobs/bolt/jobs/default_settings.py

@@ -0,0 +1,3 @@
+JOB_MIDDLEWARE: list[str] = [
+    "bolt.jobs.middleware.AppLoggerJobMiddleware",
+]

bolt-jobs/bolt/jobs/middleware.py

@@ -0,0 +1,15 @@
+from bolt.logs import app_logger
+
+
+class AppLoggerJobMiddleware:
+    def __init__(self, process_job_result):
+        self.process_job_result = process_job_result
+
+    def __call__(self, job_result):
+        app_logger.kv.context["job_request_uuid"] = str(job_result.job_request_uuid)
+        app_logger.kv.context["job_result_uuid"] = str(job_result.uuid)
+
+        self.process_job_result(job_result)
+
+        app_logger.kv.context.pop("job_request_uuid", None)
+        app_logger.kv.context.pop("job_result_uuid", None)

bolt-jobs/bolt/jobs/workers.py

@@ -5,8 +5,8 @@
 from concurrent.futures import ProcessPoolExecutor
 
 from bolt.db import transaction
-from bolt.logs import app_logger
 from bolt.signals import request_finished, request_started
+from bolt.utils.module_loading import import_string
 
 from .models import JobRequest, JobResult
 
@@ -127,13 +127,29 @@ def process_job_result(job_result_uuid):
         job_result.source,
     )
 
-    app_logger.kv.context["job_request_uuid"] = str(job_result.job_request_uuid)
-    app_logger.kv.context["job_result_uuid"] = str(job_result.uuid)
+    print("ONE")
 
-    job_result.process_job()
+    # At the core, we use the regular job model processing
+    middleware_chain = lambda job_result: job_result.process_job()
 
-    app_logger.kv.context.pop("job_request_uuid", None)
-    app_logger.kv.context.pop("job_result_uuid", None)
+    print("TWO")
+
+    from bolt.runtime import settings
+
+    for middleware_path in settings.JOB_MIDDLEWARE.reverse():
+        print(middleware_path)
+        middleware_class = import_string(middleware_path)
+        print(middleware_class)
+        middleware_instance = middleware_class(middleware_chain)
+        print(middleware_instance)
+        middleware_chain = middleware_instance
+        print(middleware_chain)
+
+    # Run the middleware chain
+
+    print("SHIT")
+    middleware_chain(job_result)
+    print("FUCK")
 
     duration = job_result.ended_at - job_result.started_at
     duration = duration.total_seconds()