Formatting pass

Author: ginwakeup

.gitignore

@@ -162,3 +162,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+examples/remote-tasks/output/app.pex

README.md

@@ -1,22 +1,17 @@
 # Salt
 <img width="994" height="340" alt="image" src="https://github.com/user-attachments/assets/19382711-28f1-44ae-8ff6-93a9915f72c7" />
 
-Salt is a Workflow/Task Scheduler system that aims to minimize dependency management for users.
+Salt is a Workflow/Task Scheduler framework that does two things:
+1. Help reducing dependency management, by providing tools to package your workflow into executables along their dependencies;
+2. Provide a Framework that can be used to define distributed DAGs.
 
-You can write dags and tasks in your favourite language (between the supported ones), with a set of unique dependencies, 
-and build them with Salt framework.
-
-After building the tasks, Salt can schedule them in isolated contexts that have all the dependencies needed to run.
+You can write dags and tasks in a supported language, define your dependencies, and package all together using Salt.
 
 ## Why Salt
-In most workflow managers (e.g. Airflow) dependency management is complex. Due to the Pythonic structure, it enforces you to use Python and take care
-of dependencies, docker images and environments, most of the time leading to complex and clashing environments
-that lead to a lot of work to keep organized.
-
-Salt tries to remove all of this. Your Workflow is self-contained with its dependencies and does not need anything but Salt framework
-to execute.
+In most workflow managers (e.g. Airflow) dependency management can get extremely complex.
 
-Tasks in the Workflows are coded using Salt framework.
+Salt simplifies the dependency management while still providing a DAG/Task Scheduling Framework. 
+Your Workflow is self-contained with its dependencies and does not need anything but Salt framework to execute.
 
 ## Getting Started
 Salt development requires `pdm` Python Package manager to be installed.

pyproject.toml

@@ -13,13 +13,13 @@ dependencies = [
     "loguru<1.0",
     "redis<7.0",
     "build<2.0",
-    "grpcio-tools",
-    "boto3",
-    "PyYaml",
-    "apscheduler",
-    "pydantic_settings",
-    "pika<2",
-    "docker"
+    "grpcio<2.0",
+    "boto3<2.0",
+    "PyYaml<7.0",
+    "apscheduler<4.0",
+    "pydantic_settings<3.0",
+    "pika<2.0",
+    "docker<8.0"
 ]
 
 requires-python = "<3.11"
@@ -33,7 +33,7 @@ salt = "salt.cli:app"
 [project.optional-dependencies]
 dev = [
     "pytest>=8.4.1",
-    "grpcio-tools"
+    "grpcio-tools<2.0"
 ]
 
 [tool.setuptools.dynamic]

src/salt/backend/scheduler/scheduler.py

@@ -52,30 +52,33 @@ 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)
-        ))
+        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()
         queue = workflow_data.get("queue", "default")
         queue = "workflows" if queue == "default" else queue
 
         channel.queue_declare(queue="workflows", durable=True)
         channel.basic_publish(
-            exchange='',
+            exchange="",
             routing_key=queue,
             body=json.dumps(workflow_data),
-            properties=pika.BasicProperties(
-                delivery_mode=2  # Make message persistent
-            )
+            properties=pika.BasicProperties(delivery_mode=2),  # Make message persistent
         )
 
         channel.close()
         rabbit.close()
 
-        logger.debug(f"Queued [{workflow_data['workflow_id']}] v{workflow_data['version']} to queue: {queue}")
-
+        logger.debug(
+            f"Queued [{workflow_data['workflow_id']}] v{workflow_data['version']} to queue: {queue}"
+        )
 
     def schedule_workflow(self, scheduler, workflow_data):
         schedule = workflow_data.get("schedule", {})
@@ -87,15 +90,15 @@ def schedule_workflow(self, scheduler, workflow_data):
                 trigger=IntervalTrigger(seconds=int(schedule["every_seconds"])),
                 id=str(uuid.uuid4()),
                 args=[workflow_data],
-                replace_existing=False
+                replace_existing=False,
             )
         elif schedule.get("type") == "cron":
             scheduler.add_job(
                 func=self.queue_workflow,
                 trigger=CronTrigger.from_crontab(schedule["cron"]),
                 id=str(uuid.uuid4()),
                 args=[workflow_data],
-                replace_existing=False
+                replace_existing=False,
             )
         else:
             print(f"Unknown schedule type for workflow {workflow_id}")

src/salt/backend/worker/settings.py

@@ -2,6 +2,7 @@
 
 from pydantic_settings import BaseSettings
 
+
 class WorkflowWorkerSettings(BaseSettings):
     queue: Optional[str] = "workflows"
 

src/salt/backend/worker/workflow_worker.py

@@ -4,6 +4,7 @@
 
 Once a Workflow Schedule is consumed by the Worker, the Workflow is executed on it and the atomic tasks will
 execute as defined."""
+
 import json
 import pathlib
 import subprocess
@@ -22,39 +23,53 @@
 
 class WorkflowWorker:
     def __init__(self):
-        self._rabbit = pika.BlockingConnection(pika.ConnectionParameters(
-            host=settings.rabbit_settings.rabbit_url,
-            credentials=pika.PlainCredentials(settings.rabbit_settings.rabbit_user,
-                                              settings.rabbit_settings.rabbit_password)
-        ))
+        self._rabbit = pika.BlockingConnection(
+            pika.ConnectionParameters(
+                host=settings.rabbit_settings.rabbit_url,
+                credentials=pika.PlainCredentials(
+                    settings.rabbit_settings.rabbit_user,
+                    settings.rabbit_settings.rabbit_password,
+                ),
+            )
+        )
 
         self._workflow_registry = workflow_registry.WorkflowRegistry()
 
     def execute_workflow(self, ch, method, properties, body):
         try:
             workflow_obj = workflow_pb2.Workflow()
-            ParseDict(json.loads(body.decode()), workflow_obj, ignore_unknown_fields=True)
+            ParseDict(
+                json.loads(body.decode()), workflow_obj, ignore_unknown_fields=True
+            )
 
             # Download and execute workflow
             with tempfile.TemporaryDirectory() as tmp_dir:
                 workflow_file_path = pathlib.Path(tmp_dir) / "workflow.pex"
-                self._workflow_registry.s3_download_workflow_binary(workflow=workflow_obj, workflow_file_path=workflow_file_path)
+                self._workflow_registry.s3_download_workflow_binary(
+                    workflow=workflow_obj, workflow_file_path=workflow_file_path
+                )
                 if not os.path.exists(workflow_file_path):
-                    logger.error(f"Workflow file was not downloaded correctly at: {workflow_file_path}")
+                    logger.error(
+                        f"Workflow file was not downloaded correctly at: {workflow_file_path}"
+                    )
 
-                logger.debug(f"Downloaded workflow binary from S3: {workflow_obj.workflow_id}")
+                logger.debug(
+                    f"Downloaded workflow binary from S3: {workflow_obj.workflow_id}"
+                )
 
                 # Execute Workflow
                 result = subprocess.run(
                     ["python3.10", str(workflow_file_path)],
                     stdout=subprocess.PIPE,
-                    stderr=subprocess.PIPE
+                    stderr=subprocess.PIPE,
                 )
 
                 logger.debug(result.stdout)
                 logger.debug(result)
 
-            ch.basic_ack(delivery_tag=method.delivery_tag)  # Acknowledge successful processing
+            ch.basic_ack(
+                delivery_tag=method.delivery_tag
+            )  # Acknowledge successful processing
         except Exception as e:
             print(f"Error processing message: {e}")
             ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
@@ -66,10 +81,13 @@ def loop(self):
         channel.queue_declare(queue=queue_name, durable=True)
 
         channel.basic_qos(prefetch_count=1)  # Ensure fair dispatch
-        channel.basic_consume(queue=queue_name, on_message_callback=self.execute_workflow)
+        channel.basic_consume(
+            queue=queue_name, on_message_callback=self.execute_workflow
+        )
 
         print(" [*] Waiting for messages. To exit press CTRL+C")
         channel.start_consuming()
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     WorkflowWorker().loop()

src/salt/backend/workflow_service/utils.py

@@ -47,7 +47,9 @@ def request_generator(workflow: Workflow, file_path: Path):
         salt_data = yaml.safe_load(f)
     wf = ParseDict(salt_data, Workflow())
 
-    with grpc.insecure_channel(f"{workflow_service_settings.workflow_service_url}:{workflow_service_settings.workflow_service_port}") as channel:
+    with grpc.insecure_channel(
+        f"{workflow_service_settings.workflow_service_url}:{workflow_service_settings.workflow_service_port}"
+    ) as channel:
         response = WorkflowServiceStub(channel=channel).RegisterWorkflow(
             request_generator(wf, pex.pop())
         )
@@ -58,6 +60,7 @@ def get_workflow_binary():
     """Given workflow metadata, downloads the binary from S3."""
     pass
 
+
 if __name__ == "__main__":
     import pathlib
 

src/salt/backend/workflow_service/workflow_pb2.py

@@ -13,14 +13,27 @@ class Trigger(_message.Message):
     def __init__(self, type: _Optional[str] = ...) -> None: ...
 
 class Workflow(_message.Message):
-    __slots__ = ("workflow_id", "chunk", "version", "paused", "owner", "schedule", "triggers", "queue")
+    __slots__ = (
+        "workflow_id",
+        "chunk",
+        "version",
+        "paused",
+        "owner",
+        "schedule",
+        "triggers",
+        "queue",
+    )
+
     class ScheduleEntry(_message.Message):
         __slots__ = ("key", "value")
         KEY_FIELD_NUMBER: _ClassVar[int]
         VALUE_FIELD_NUMBER: _ClassVar[int]
         key: str
         value: str
-        def __init__(self, key: _Optional[str] = ..., value: _Optional[str] = ...) -> None: ...
+        def __init__(
+            self, key: _Optional[str] = ..., value: _Optional[str] = ...
+        ) -> None: ...
+
     WORKFLOW_ID_FIELD_NUMBER: _ClassVar[int]
     CHUNK_FIELD_NUMBER: _ClassVar[int]
     VERSION_FIELD_NUMBER: _ClassVar[int]
@@ -37,7 +50,17 @@ class Workflow(_message.Message):
     schedule: _containers.ScalarMap[str, str]
     triggers: _containers.RepeatedCompositeFieldContainer[Trigger]
     queue: str
-    def __init__(self, workflow_id: _Optional[str] = ..., chunk: _Optional[bytes] = ..., version: _Optional[str] = ..., paused: bool = ..., owner: _Optional[str] = ..., schedule: _Optional[_Mapping[str, str]] = ..., triggers: _Optional[_Iterable[_Union[Trigger, _Mapping]]] = ..., queue: _Optional[str] = ...) -> None: ...
+    def __init__(
+        self,
+        workflow_id: _Optional[str] = ...,
+        chunk: _Optional[bytes] = ...,
+        version: _Optional[str] = ...,
+        paused: bool = ...,
+        owner: _Optional[str] = ...,
+        schedule: _Optional[_Mapping[str, str]] = ...,
+        triggers: _Optional[_Iterable[_Union[Trigger, _Mapping]]] = ...,
+        queue: _Optional[str] = ...,
+    ) -> None: ...
 
 class Result(_message.Message):
     __slots__ = ("success", "message")