refactor: flows to pipelines

Author: jsbroks

datatorch/__init__.py

@@ -7,9 +7,10 @@
 __all__ = ["ApiClient", "get_inputs"]
 
 
-def get_inputs() -> dict:
+def get_inputs(key: str = None) -> dict:
     try:
-        return json.loads(sys.argv[-1])
+        values = json.loads(sys.argv[-1])
+        return values.get(key) if key else values
     except Exception as e:
         return {}
 

datatorch/agent/__init__.py

@@ -1,7 +1,6 @@
 import asyncio
 import logging
 import click
-import sys
 import os
 
 from logging.handlers import RotatingFileHandler
@@ -69,7 +68,7 @@ async def _exit_tasks() -> None:
     tasks = [
         task
         for task in asyncio.Task.all_tasks()
-        if task is not asyncio.tasks.Task.current_task()
+        if task is not asyncio.tasks.Task.current_task()  # type: ignore
     ]
 
     for task in tasks:

datatorch/agent/agent.py

@@ -5,13 +5,13 @@
 
 from typing import List
 from .client import AgentApiClient
-from .flows.template import Variables
+from .pipelines.template import Variables
 from .log_handler import AgentAPIHandler
 from .monitoring import AgentSystemStats
 from .directory import agent_directory
 
 from gql.client import AsyncClientSession  # type: ignore
-from datatorch.agent.flows import Job
+from datatorch.agent.pipelines import Job
 
 
 logger = logging.getLogger(__name__)
@@ -61,10 +61,10 @@ async def _run_job(self, job):
         job_name = job.get("name")
         job_steps = job.get("steps")
 
-        flow_run = job.get("run")
-        flow_config = flow_run.get("config")
+        run = job.get("run")
+        run_config = run.get("config")
 
-        job_config = flow_config.get("jobs").get(job_name)
+        job_config = run_config.get("jobs").get(job_name)
         job_config["id"] = job_id
         job_config["name"] = job_name
 
@@ -87,7 +87,7 @@ async def _run_job(self, job):
             if step.get("id") == None:
                 raise ValueError(f"No ID found for step {step.get('action')}.")
 
-        variables = Variables(flow_run)
+        variables = Variables(run)
 
         try:
             logger.info(f"Starting {job_name} {job_id}")

datatorch/agent/client.py

@@ -28,7 +28,7 @@ def agent_jobs(self):
                         text
                         config
                         runNumber
-                        flow {
+                        pipeline {
                             id
                             projectId
                             creatorId
@@ -124,7 +124,7 @@ def update_step(self, values: dict):
                 $startedAt: DateTime
                 $finishedAt: DateTime
             ) {
-                updateFlowStep(
+                updatePipelineStep(
                     id: $id
                     input: {
                         status: $status
@@ -142,8 +142,8 @@ def update_step(self, values: dict):
     async def update_job(self, values: dict):
         # fmt: off
         mutate = """
-            mutation UpdateFlowJob($id: ID!, $status: String) {
-                updateFlowJobRun(
+            mutation UpdateJob($id: ID!, $status: String) {
+                updatePipelineJobRun(
                     id: $id,
                     input: { status: $status }
                 )
@@ -155,8 +155,8 @@ async def update_job(self, values: dict):
     def upload_step_logs(self, step_id: str, logs: List[Log]):
         # fmt: off
         mutate = """
-            mutation UploadStepLogs($id: ID!, $logs: [CreateFlowStepLog!]!) {
-                createFlowStepLog(stepId: $id, logs: $logs)
+            mutation UploadStepLogs($id: ID!, $logs: [CreatePipelineStepLog!]!) {
+                createPipelineStepLog(stepId: $id, logs: $logs)
             }
         """
         # fmt: on

datatorch/agent/flows/__init__.py

@@ -0,0 +1,4 @@
+from .pipeline import Pipeline
+from .job import Job
+
+__all__ = ["Pipeline", "Job"]

datatorch/agent/flows/flow.py

@@ -1,4 +1,4 @@
-from datatorch.agent.flows.template import Variables
+from datatorch.agent.pipelines.template import Variables
 import os
 import yaml
 import logging

datatorch/agent/pipelines/__init__.py

@@ -28,17 +28,15 @@ def __init__(self, config: dict, agent: "Agent" = None):
                 yaml.dump(self.config, yaml_config, default_flow_style=False)
 
     async def update(self, status: str) -> None:
-        if self.agent is None:
+        if self.agent is None or self.id is None:
             return None
         variables = {"id": self.id, "status": status}
         await self.agent.api.update_job(variables)
 
     async def run(self, variables: Variables):
         """ Runs each step of the job. """
         steps = Step.from_dict_list(self.config.get("steps", []), job=self)
-        inputs = {}
         await self.update("RUNNING")
-
         variables.set_job(self)
 
         for step in steps:

datatorch/agent/flows/action/__init__.py datatorch/agent/pipelines/action/__init__.py

@@ -0,0 +1,41 @@
+import os
+import yaml
+import logging
+
+from typing import Union, TYPE_CHECKING
+from .job import Job
+
+
+if TYPE_CHECKING:
+    from ..agent import Agent
+
+
+logger = logging.getLogger("datatorch.agent.pipeline")
+
+
+class Pipeline(object):
+    @classmethod
+    def from_yaml(cls, path, agent: "Agent" = None):
+        with open(path, "r") as config_file:
+            config = yaml.load(config_file, Loader=yaml.FullLoader)
+        config["name"] = config.get("name", os.path.splitext(os.path.basename(path))[0])
+        return cls.from_config(config, agent)
+
+    @classmethod
+    def from_config(cls, config: Union[str, dict], agent: "Agent" = None):
+        """ Creates a pipeline from a config file. """
+        if isinstance(config, str):
+            cf = yaml.load(config, Loader=yaml.FullLoader)
+        else:
+            cf = config
+        return cls(cf, agent=agent)
+
+    def __init__(self, config: dict, agent: "Agent" = None):
+        self.name = config.get("name")
+        self.config = config
+        self.agent = agent
+
+    async def run(self, job_config: dict):
+        """ Runs a job. """
+        # await Job(job_config, agent=self.agent).run()
+        pass

datatorch/agent/flows/action/action.py datatorch/agent/pipelines/action/action.py

@@ -12,7 +12,7 @@
 
 global_variables = {
     "machine": {
-        "name": platform.machine(),
+        "name": platform.node(),
         "os": platform.system(),
         "version": platform.version(),
     },
@@ -24,28 +24,28 @@
 
 
 class Variables(object):
-    def __init__(self, flow_run: dict):
+    def __init__(self, run: dict):
         self.variables = {"variable": {}, "input": {}}
-        flow = flow_run.get("flow")
+        pipeline = run.get("pipeline")
         self.set(
-            "flow",
+            "pipeline",
             {
-                "id": flow.get("id"),
-                "name": flow.get("name"),
-                "creatorId": flow.get("creatorId"),
-                "projectId": flow.get("projectId"),
-                "lastRunNumber": flow.get("lastRunNumber"),
+                "id": pipeline.get("id"),
+                "name": pipeline.get("name"),
+                "creatorId": pipeline.get("creatorId"),
+                "projectId": pipeline.get("projectId"),
+                "lastRunNumber": pipeline.get("lastRunNumber"),
             },
         )
 
         self.set(
             "run",
             {
-                "id": flow_run.get("id"),
-                "name": flow_run.get("name"),
-                "config": flow_run.get("config"),
-                "createdAt": flow_run.get("createdAt"),
-                "runNumber": flow_run.get("runNumber"),
+                "id": run.get("id"),
+                "name": run.get("name"),
+                "config": run.get("config"),
+                "createdAt": run.get("createdAt"),
+                "runNumber": run.get("runNumber"),
             },
         )
         print(self.variables)
@@ -78,7 +78,12 @@ def merge(self, section: str, variables: dict):
         self.variables[section] = {**self.variables[section], **variables}
 
     def render(self, string: str):
-        tp = Template(string)
+        tp = Template(
+            string,
+            block_start_string="${%",
+            variable_start_string="${{",
+            comment_start_string="${#",
+        )
         return tp.render({**global_variables, **self.variables})
 
     @property

datatorch/agent/flows/action/config.py datatorch/agent/pipelines/action/config.py

@@ -101,7 +101,7 @@ def datasets(self) -> List[Dataset]:
             params={"projectId": self.id},
         )
 
-    def files(self, where: Where = None, limit=500, page=1):
+    def files(self, where: Where = None, limit=500, page=1) -> List[File]:
         if where is None:
             where = Where()
         return self.client.query_to_class(

datatorch/agent/flows/job/__init__.py datatorch/agent/pipelines/job/__init__.py

@@ -23,11 +23,11 @@ def xywh(cls, x, y, width, height):
     @property
     def top_left(self) -> Point2D:
         """ Top-left point of the box """
-        return [self.x, self.y]
+        return (self.x, self.y)
 
     @property
     def bottom_right(self) -> Point2D:
-        return [self.x + self.width, self.y + self.height]
+        return (self.x + self.width, self.y + self.height)
 
     @property
     def size(self) -> float:

datatorch/agent/flows/job/job.py datatorch/agent/pipelines/job/job.py

@@ -6,7 +6,7 @@
 import json
 
 from datatorch.agent import logger
-from datatorch.agent.flows.action import Action
+from datatorch.agent.pipelines.action import Action
 
 
 input_types = {"string": str, "boolean": bool, "number": float}

datatorch/agent/pipelines/pipeline.py

@@ -9,7 +9,7 @@
 from .main.version import version
 
 from .agent import agent
-from .flow import flow
+from .pipeline import pipeline
 from .action import action
 
 
@@ -24,6 +24,6 @@ def main():
 main.add_command(version)
 main.add_command(package_upgrade)
 
-main.add_command(flow)
+main.add_command(pipeline)
 main.add_command(agent)
 main.add_command(action)

datatorch/agent/flows/runner/__init__.py datatorch/agent/pipelines/runner/__init__.py

@@ -6,10 +6,10 @@
 
 
 @click.group()
-def flow():
+def pipeline():
     pass
 
 
-flow.add_command(run)
-flow.add_command(generate)
-flow.add_command(upload)
+pipeline.add_command(run)
+pipeline.add_command(generate)
+pipeline.add_command(upload)

datatorch/agent/flows/runner/command.py datatorch/agent/pipelines/runner/command.py

@@ -2,20 +2,16 @@
 import yaml
 import asyncio
 
-from datatorch.agent.flows import Flow, Job
+from datatorch.agent.pipelines import Pipeline, Job
 from datatorch.agent import setup_logging
 
 
-class FlowRun(object):
-    pass
-
-
-@click.command(help="Runs a flow yaml file on local machine.")
+@click.command(help="Runs a pipeline yaml file on local machine.")
 @click.argument("path", type=click.Path(exists=True))
 def run(path):
     setup_logging()
 
-    async def run_jobs(flow: Flow):
+    async def run_jobs(flow: Pipeline):
         """ Run tasks in parallel """
 
         tasks = []
@@ -26,4 +22,4 @@ async def run_jobs(flow: Flow):
         await asyncio.wait(tasks)
 
     loop = asyncio.get_event_loop()
-    loop.run_until_complete(run_jobs(Flow.from_yaml(path)))
+    loop.run_until_complete(run_jobs(Pipeline.from_yaml(path)))