ability to trigger existing workflows

metaflow/metaflow_config.py

@@ -410,6 +410,13 @@
 )
 
 
+###
+# From Deployment configuration
+###
+# From Deployment Configuration for DeployedFlow
+FROM_DEPLOYMENT_IMPL = from_conf("FROM_DEPLOYMENT_IMPL", "argo-workflows")
+
+
 ###
 # Conda configuration
 ###

metaflow/plugins/__init__.py

@@ -156,6 +156,12 @@ def get_plugin_cli():
     return resolve_plugins("cli")
 
 
+# just a normal dictionary, not a plugin since there is no class..
+FROM_DEPLOYMENT_PROVIDERS = {
+    "argo-workflows": "metaflow.plugins.argo.argo_workflows_deployer",
+}
+
+
 STEP_DECORATORS = resolve_plugins("step_decorator")
 FLOW_DECORATORS = resolve_plugins("flow_decorator")
 ENVIRONMENTS = resolve_plugins("environment")

metaflow/plugins/argo/argo_workflows_deployer.py

@@ -1,9 +1,15 @@
 import sys
+import json
 import tempfile
 from typing import Optional, ClassVar
 
+from metaflow.client.core import get_metadata
+from metaflow.exception import MetaflowException
+from metaflow.plugins.argo.argo_client import ArgoClient
+from metaflow.metaflow_config import KUBERNETES_NAMESPACE
 from metaflow.plugins.argo.argo_workflows import ArgoWorkflows
 from metaflow.runner.deployer import (
+    Deployer,
     DeployerImpl,
     DeployedFlow,
     TriggeredRun,
@@ -12,6 +18,100 @@
 )
 
 
+def generate_fake_flow_file_contents(
+    flow_name: str, param_info: dict, project_name: Optional[str] = None
+):
+    params_code = ""
+    for _, param_details in param_info.items():
+        param_name = param_details["name"]
+        param_type = param_details["type"]
+        param_help = param_details["description"]
+        param_required = param_details["is_required"]
+
+        if param_type == "JSON":
+            params_code += f"    {param_name} = Parameter('{param_name}', type=JSONType, help='{param_help}', required={param_required})\n"
+        elif param_type == "FilePath":
+            is_text = param_details.get("is_text", True)
+            encoding = param_details.get("encoding", "utf-8")
+            params_code += f"    {param_name} = IncludeFile('{param_name}', is_text={is_text}, encoding='{encoding}', help='{param_help}', required={param_required})\n"
+        else:
+            params_code += f"    {param_name} = Parameter('{param_name}', type={param_type}, help='{param_help}', required={param_required})\n"
+
+    project_decorator = f"@project(name='{project_name}')\n" if project_name else ""
+
+    contents = f"""\
+from metaflow import FlowSpec, Parameter, IncludeFile, JSONType, step, project
+{project_decorator}class {flow_name}(FlowSpec):
+{params_code}
+    @step
+    def start(self):
+        self.next(self.end)
+    @step
+    def end(self):
+        pass
+if __name__ == '__main__':
+    {flow_name}()
+"""
+    return contents
+
+
+def from_deployment(identifier: str, metadata: Optional[str] = None):
+    client = ArgoClient(namespace=KUBERNETES_NAMESPACE)
+    workflow_template = client.get_workflow_template(identifier)
+
+    if workflow_template is None:
+        raise MetaflowException("No deployed flow found for: %s" % identifier)
+
+    metadata_annotations = workflow_template.get("metadata", {}).get("annotations", {})
+
+    flow_name = metadata_annotations.get("metaflow/flow_name", "")
+    username = metadata_annotations.get("metaflow/owner", "")
+    parameters = json.loads(metadata_annotations.get("metaflow/parameters", {}))
+
+    # these two only exist if @project decorator is used..
+    branch_name = metadata_annotations.get("metaflow/branch_name", None)
+    project_name = metadata_annotations.get("metaflow/project_name", None)
+
+    project_kwargs = {}
+    if branch_name is not None:
+        if branch_name.startswith("prod."):
+            project_kwargs["production"] = True
+            project_kwargs["branch"] = branch_name[len("prod.") :]
+        elif branch_name.startswith("test."):
+            project_kwargs["branch"] = branch_name[len("test.") :]
+        elif branch_name == "prod":
+            project_kwargs["production"] = True
+
+    fake_flow_file_contents = generate_fake_flow_file_contents(
+        flow_name=flow_name, param_info=parameters, project_name=project_name
+    )
+
+    with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as fake_flow_file:
+        with open(fake_flow_file.name, "w") as fp:
+            fp.write(fake_flow_file_contents)
+
+        if branch_name is not None:
+            d = Deployer(
+                fake_flow_file.name, env={"METAFLOW_USER": username}, **project_kwargs
+            ).argo_workflows()
+        else:
+            d = Deployer(
+                fake_flow_file.name, env={"METAFLOW_USER": username}
+            ).argo_workflows(name=identifier)
+
+        d.name = identifier
+        d.flow_name = flow_name
+        if metadata is None:
+            d.metadata = get_metadata()
+        else:
+            d.metadata = metadata
+
+    df = DeployedFlow(deployer=d)
+    d._enrich_deployed_flow(df)
+
+    return df
+
+
 def suspend(instance: TriggeredRun, **kwargs):
     """
     Suspend the running workflow.

metaflow/runner/deployer.py

@@ -214,7 +214,43 @@ def run(self):
             return None
 
 
-class DeployedFlow(object):
+class LazyDeploymentMethod:
+    def __init__(self, module_path, func_name):
+        self.module_path = module_path
+        self.func_name = func_name
+        self.func = None
+
+    def __call__(self, *args, **kwargs):
+        if self.func is None:
+            module = importlib.import_module(self.module_path)
+            self.func = getattr(module, self.func_name)
+        return self.func(*args, **kwargs)
+
+
+class DeploymentMethodsMeta(type):
+    from metaflow.plugins import FROM_DEPLOYMENT_PROVIDERS
+    from metaflow.metaflow_config import FROM_DEPLOYMENT_IMPL
+
+    def __new__(mcs, name, bases, dct):
+        cls = super().__new__(mcs, name, bases, dct)
+
+        def from_deployment(identifier, metadata=None, impl=None):
+            if impl is None:
+                impl = mcs.FROM_DEPLOYMENT_IMPL
+
+            if impl not in mcs.FROM_DEPLOYMENT_PROVIDERS:
+                raise ValueError("This method is not available for: %s" % impl)
+
+            module_path = mcs.FROM_DEPLOYMENT_PROVIDERS[impl]
+            lazy_method = LazyDeploymentMethod(module_path, "from_deployment")
+            return lazy_method(identifier=identifier, metadata=metadata)
+
+        setattr(cls, "from_deployment", staticmethod(from_deployment))
+
+        return cls
+
+
+class DeployedFlow(metaclass=DeploymentMethodsMeta):
     """
     DeployedFlow class represents a flow that has been deployed.
 
@@ -226,6 +262,9 @@ class DeployedFlow(object):
 
     def __init__(self, deployer: "DeployerImpl"):
         self.deployer = deployer
+        self.name = self.deployer.name
+        self.flow_name = self.deployer.flow_name
+        self.metadata = self.deployer.metadata
 
     def _enrich_object(self, env):
         """