Rabbit settings

Author: ginwakeup

src/salt/backend/scheduler/scheduler.py

@@ -18,6 +18,7 @@
 class Scheduler:
     def __init__(self):
         self._redis = settings.redis_settings.redis_client
+        self._rabbit = settings.rabbit_settings.rabbit_client
 
     def loop(self):
         sched = BackgroundScheduler()
@@ -47,16 +48,7 @@ def queue_workflow(self, workflow_data):
         TODO The handling of KILL signal (e.g. to stop the workflow) is handled using Kafka.
           Once the job is picked by a worker, the worker starts listening for KILL signals through Kafka or similar.
         """
-        rabbit = pika.BlockingConnection(
-            pika.ConnectionParameters(
-                host=settings.rabbit_settings.rabbit_url,
-                credentials=pika.PlainCredentials(
-                    settings.rabbit_settings.rabbit_user,
-                    settings.rabbit_settings.rabbit_password,
-                ),
-            )
-        )
-        channel = rabbit.channel()
+        channel = self._rabbit.channel()
         queue = workflow_data.get("queue", "default")
         queue = "workflows" if queue == "default" else queue
 

src/salt/settings.py

@@ -1,5 +1,6 @@
 import redis
 import boto3
+import pika
 from botocore import client
 
 from typing import Optional
@@ -28,6 +29,17 @@ class RabbitSettings(BaseSettings):
     rabbit_user: Optional[str] = "admin"
     rabbit_password: Optional[str] = "admin"
 
+    @computed_field(return_type=pika.BlockingConnection)
+    def rabbit_client(self):
+        pika.BlockingConnection(
+            pika.ConnectionParameters(
+                host=self.rabbit_url,
+                credentials=pika.PlainCredentials(
+                    self.rabbit_user,
+                    self.rabbit_password,
+                ),
+            )
+        )
 
 class S3Settings(BaseSettings):
     s3_endpoint_url: Optional[str] = "http://localhost:9000"