Add run-operation to click CLI

core/dbt/cli/main.py

@@ -15,6 +15,7 @@
 from dbt.task.run import RunTask
 from dbt.task.test import TestTask
 from dbt.task.snapshot import SnapshotTask
+from dbt.task.run_operation import RunOperationTask
 
 
 # CLI invocation
@@ -349,6 +350,7 @@ def run(ctx, **kwargs):
 
 # dbt run operation
 @cli.command("run-operation")
+@click.argument("macro")
 @click.pass_context
 @p.args
 @p.profile
@@ -357,10 +359,16 @@ def run(ctx, **kwargs):
 @p.target
 @p.vars
 @requires.preflight
+@requires.profile
+@requires.project
 def run_operation(ctx, **kwargs):
     """Run the named macro with any supplied arguments."""
-    click.echo(f"`{inspect.stack()[0][3]}` called\n flags: {ctx.obj['flags']}")
-    return None, True
+    config = RuntimeConfig.from_parts(ctx.obj["project"], ctx.obj["profile"], ctx.obj["flags"])
+    task = RunOperationTask(ctx.obj["flags"], config)
+
+    results = task.run()
+    success = task.interpret_results(results)
+    return results, success
 
 
 # dbt seed

core/dbt/task/run_operation.py

@@ -1,5 +1,4 @@
 from datetime import datetime
-from typing import Dict, Any
 import traceback
 
 import agate
@@ -8,7 +7,6 @@
 
 import dbt.exceptions
 from dbt.adapters.factory import get_adapter
-from dbt.config.utils import parse_cli_vars
 from dbt.contracts.results import RunOperationResultsArtifact
 from dbt.exceptions import InternalException
 from dbt.events.functions import fire_event
@@ -29,14 +27,6 @@ def _get_macro_parts(self):
 
         return package_name, macro_name
 
-    def _get_kwargs(self) -> Dict[str, Any]:
-        # N.B. parse_cli_vars is embedded into the param when using click.
-        # replace this with:
-        # return self.args.args
-        # when this task is refactored for click
-        # or remove the function completely as it's basically a noop
-        return parse_cli_vars(self.args.args)
-
     def compile_manifest(self) -> None:
         if self.manifest is None:
             raise InternalException("manifest was None in compile_manifest")
@@ -45,7 +35,7 @@ def _run_unsafe(self) -> agate.Table:
         adapter = get_adapter(self.config)
 
         package_name, macro_name = self._get_macro_parts()
-        macro_kwargs = self._get_kwargs()
+        macro_kwargs = self.args.args
 
         with adapter.connection_named("macro_{}".format(macro_name)):
             adapter.clear_transaction()