Merge pull request #26 from quantile-development/refactor

Multiple refactors
quantile-development · Feb 18, 2023 · 0517c3c · 0517c3c
2 parents 3d44eaf + c3f75b3
commit 0517c3c
Show file tree

Hide file tree

Showing 20 changed files with 743 additions and 773 deletions.
diff --git a/README.md b/README.md
@@ -49,6 +49,21 @@ ops:
         TAP_SMOKE_TEST_STREAMS: '[{"stream_name": "new-stream", "input_filename": "demo.json"}]'
 ```
 
+An example of running an arbitrary Meltano command.
+
+```python
+from dagster import repository, job
+from dagster_meltano import meltano_resource, meltano_command_op
+
+@job(resource_defs={"meltano": meltano_resource})
+def meltano_command_job():
+    meltano_command_op("install loader tap-smoke-test")()
+
+@repository()
+def repository():
+    return [meltano_command_job]
+```
+
 ## Development using VSCode
 
 1. Open this repository in Visual Studio Code.

diff --git a/dagster_meltano/__init__.py b/dagster_meltano/__init__.py
@@ -5,4 +5,4 @@
     load_jobs_from_meltano_project,
 )
 from dagster_meltano.meltano_resource import MeltanoResource, meltano_resource
-from dagster_meltano.ops import meltano_install_op, meltano_run_op
+from dagster_meltano.ops import meltano_command_op, meltano_install_op, meltano_run_op
diff --git a/dagster_meltano/assets.py b/dagster_meltano/assets.py
diff --git a/dagster_meltano/exceptions.py b/dagster_meltano/exceptions.py
@@ -0,0 +1,2 @@
+class MeltanoCommandError(Exception):
+    pass
diff --git a/dagster_meltano/job.py b/dagster_meltano/job.py
@@ -14,20 +14,14 @@
     op,
 )
 
-from dagster_meltano.meltano_invoker import MeltanoInvoker
 from dagster_meltano.ops import meltano_run_op as meltano_run_op_factory
 from dagster_meltano.utils import generate_dagster_name
 
 
 class Job:
-    def __init__(
-        self,
-        meltano_job: dict,
-        meltano_invoker: MeltanoInvoker,
-    ) -> None:
+    def __init__(self, meltano_job: dict) -> None:
         self.name = meltano_job["job_name"]
         self.tasks = meltano_job["tasks"]
-        self.meltano_invoker = meltano_invoker
 
     @property
     def dagster_name(self) -> str:
@@ -54,44 +48,10 @@ def dagster_job():
                 meltano_run_op = meltano_run_op_factory(task)
                 current_task_contains_tap = self.task_contains_tap(task)
 
-                # # If the task does not contain a tap, we generate a new layer
-                # if not self.task_contains_tap(task):
-                #     op_layers.append([])
-
                 if not current_task_contains_tap or not previous_task_contains_tap:
                     op_layers.append([])
 
-                # if not current_task_contains_tap and previous_task_contains_tap:
-
-                # op_layers.append([])
-
-                # # When we are in the first layer
-                # if len(op_layers) == 1:
-                #     op_layers[-1].append(meltano_run_op())
-                #     continue
-
-                op_layers[-1].append(meltano_run_op(op_layers[-2]))
+                op_layers[-1].append(meltano_run_op(after=op_layers[-2]))
                 previous_task_contains_tap = current_task_contains_tap
 
-            # logging.warning(op_layers)
-
-            # for op_layer_index, op_layer in enumerate(op_layers)[1:]:
-            #     previous_layer = op_layers[op_layer_index - 1]
-
-            #     for
-            #     [meltano_run_op(previous_layer) for meltano_run_op in op_layer]
-
-            # meltano_run_op()
-
-            # meltano_run = meltano_run_op(command=task)
-
-            # if previous_done:
-            #     previous_done = meltano_run(previous_done)
-            # else:
-            #     previous_done = meltano_run()
-
-            # if self.task_contains_tap(task):
-            #     previous_done.append(previous_done)
-            #     continue
-
         return dagster_job
diff --git a/dagster_meltano/log_processing.py b/dagster_meltano/log_processing.py
diff --git a/dagster_meltano/log_processing/__init__.py b/dagster_meltano/log_processing/__init__.py
diff --git a/dagster_meltano/log_processing/json_processor.py b/dagster_meltano/log_processing/json_processor.py
diff --git a/dagster_meltano/log_processing/metadata_processor.py b/dagster_meltano/log_processing/metadata_processor.py
diff --git a/dagster_meltano/log_processing/passthrough_processor.py b/dagster_meltano/log_processing/passthrough_processor.py
diff --git a/dagster_meltano/logging.yaml b/dagster_meltano/logging.yaml
@@ -20,7 +20,7 @@ handlers:
   console:
     class: logging.StreamHandler
     level: INFO
-    formatter: json
+    formatter: default
     stream: "ext://sys.stderr"
 
 root: