Centralize connection creation

Author: ginwakeup

examples/workflows_python/__init__.py

@@ -0,0 +1,23 @@
+"""This module shows a simple workflow definition in Python."""
+from salt.framework import workflow, task
+
+
+class ExampleWorkflow(workflow.Workflow):
+    @task.task
+    def my_first_task(self):
+        print("Hello!")
+
+    @task.task
+    def my_second_task(self):
+        print("World!")
+
+    def graph(self):
+        self.my_first_task >> self.my_second_task
+
+
+def main():
+    ExampleWorkflow().run()
+
+
+if __name__ == "__main__":
+    ExampleWorkflow().run()

examples/workflows_python/simple_workflow.py

@@ -0,0 +1,23 @@
+"""This module shows a Workflow with Tasks that return values."""
+from salt.framework import workflow, task
+
+
+class ExampleWorkflow(workflow.Workflow):
+    @task.task
+    def my_first_task(self):
+        print("Hello!")
+
+    @task.task
+    def my_second_task(self):
+        print("World!")
+
+    def graph(self):
+        self.my_first_task >> self.my_second_task
+
+
+def main():
+    ExampleWorkflow().run()
+
+
+if __name__ == "__main__":
+    ExampleWorkflow().run()

examples/workflows_python/tasks_with_return_values.py

@@ -4,7 +4,6 @@
 import json
 import uuid
 
-import redis
 import pika
 
 from apscheduler.schedulers.background import BackgroundScheduler
@@ -18,11 +17,7 @@
 
 class Scheduler:
     def __init__(self):
-        self._redis = redis.Redis(
-            host=settings.redis_settings.redis_url,
-            port=settings.redis_settings.redis_port,
-            db=settings.redis_settings.redis_workflow_db,
-        )
+        self._redis = settings.redis_settings.redis_client
 
     def loop(self):
         sched = BackgroundScheduler()

src/salt/backend/scheduler/scheduler.py

@@ -41,3 +41,19 @@ def my_dynamic_task
 ```
 
 Only the first one will be executed on the Worker that picked the workflow, while the others will be distributed.
+
+# Task & Workflow Execution
+The Workflow Worker is the component that takes care of initializing, monitoring and concluding the execution of an entire workflow.
+
+Tasks will be queued by the Workflow Worker that picked the workflow, using the information defined by the user:
+- type of task
+- task queues
+- arguments
+and so on.
+
+Once the Workflow Worker iterates through all the defined tasks in the Workflow and builds the DAG locally, it will start
+queuing them to remote queues (remote tasks) or run them locally.
+
+Workflows themselves can have queues, so users can define totally isolated fleets of Workflow Workers that do not affect each other.
+
+Tasks are executed on Workflow Workers that have been configured to listen for updates on a specific Queue.

src/salt/backend/worker/README.md

@@ -22,6 +22,7 @@
 
 
 class WorkflowWorker:
+
     def __init__(self):
         self._rabbit = pika.BlockingConnection(
             pika.ConnectionParameters(

src/salt/backend/worker/workflow_worker.py

@@ -1,10 +1,6 @@
 """The workflow registry class exposes utilities to manage Workflows in the workflow table."""
-
 import io
 import pathlib
-
-import redis
-import boto3
 import json
 import uuid
 
@@ -21,17 +17,8 @@
 
 class WorkflowRegistry:
     def __init__(self):
-        self._redis = redis.Redis(
-            host=settings.redis_settings.redis_url,
-            port=settings.redis_settings.redis_port,
-            db=settings.redis_settings.redis_workflow_db,
-        )
-        self._s3 = boto3.client(
-            "s3",
-            endpoint_url=settings.s3_settings.s3_endpoint_url,
-            aws_access_key_id=settings.s3_settings.s3_aws_access_key_id,
-            aws_secret_access_key=settings.s3_settings.s3_aws_secret_access_key,
-        )
+        self._redis = settings.redis_settings.redis_client
+        self._s3 = settings.s3_settings.s3_client
 
     def register_workflow(self, request_iterator):
         """This method registers a user uploaded workflow in the backend registry."""

src/salt/backend/workflow_service/workflow_registry.py

@@ -1,5 +1,10 @@
+import redis
+import boto3
+from botocore import client
+
 from typing import Optional
 
+from pydantic import computed_field
 from pydantic_settings import BaseSettings
 
 
@@ -8,6 +13,14 @@ class Redis(BaseSettings):
     redis_port: Optional[int] = "6379"
     redis_workflow_db: Optional[int] = 0
 
+    @computed_field(return_type=redis.Redis)
+    def redis_client(self):
+        return redis.Redis(
+            host=self.redis_url,
+            port=self.redis_port,
+            db=self.redis_workflow_db,
+        )
+
 
 class RabbitSettings(BaseSettings):
     rabbit_url: Optional[str] = "localhost"
@@ -21,6 +34,14 @@ class S3Settings(BaseSettings):
     s3_aws_access_key_id: Optional[str] = "miniadmin"
     s3_aws_secret_access_key: Optional[str] = "miniadmin"
 
+    @computed_field(return_type=client.BaseClient)
+    def s3_client(self):
+        return boto3.client(
+            "s3",
+            endpoint_url=self.s3_endpoint_url,
+            aws_access_key_id=self.s3_aws_access_key_id,
+            aws_secret_access_key=self.s3_aws_secret_access_key,
+        )
 
 class WorkflowServiceSettings(BaseSettings):
     workflow_service_url: Optional[str] = "localhost"