[WIP] Feat/user decorators v2

metaflow/__init__.py

@@ -104,7 +104,13 @@ class and related decorators.
 from .parameters import Parameter, JSONTypeClass, JSONType
 
 from .user_configs.config_parameters import Config, ConfigValue, config_expr
-from .user_configs.config_decorators import CustomFlowDecorator, CustomStepDecorator
+from .user_decorators.user_step_decorator import (
+    UserStepDecorator,
+    StepMutator,
+    user_step_decorator,
+    USER_SKIP_STEP,
+)
+from .user_decorators.user_flow_decorator import FlowMutator
 
 # data layer
 # For historical reasons, we make metaflow.plugins.datatools accessible as

metaflow/cli.py

@@ -153,8 +153,13 @@ def check(obj, warnings=False):
 def show(obj):
     echo_always("\n%s" % obj.graph.doc)
     for node_name in obj.graph.sorted_nodes:
+        echo_always("")
         node = obj.graph[node_name]
-        echo_always("\nStep *%s*" % node.name, err=False)
+        for deco in node.decorators:
+            echo_always("@%s" % deco.name, err=False)
+        for deco in node.wrappers:
+            echo_always("@%s" % deco.decorator_name, err=False)
+        echo_always("Step *%s*" % node.name, err=False)
         echo_always(node.doc if node.doc else "?", indent=True, err=False)
         if node.type != "end":
             echo_always(
@@ -442,6 +447,10 @@ def start(
         # be raised. For resume, since we ignore those options, we ignore the error.
         raise ctx.obj.delayed_config_exception
 
+    # Init all values in the config decorators and then process them
+    for decorator in ctx.obj.flow._flow_state.get(_FlowState.CONFIG_DECORATORS, []):
+        decorator.external_init()
+
     new_cls = ctx.obj.flow._process_config_decorators(config_options)
     if new_cls:
         ctx.obj.flow = new_cls(use_cli=False)

metaflow/datastore/task_datastore.py

@@ -99,7 +99,6 @@ def __init__(
         mode="r",
         allow_not_done=False,
     ):
-
         self._storage_impl = flow_datastore._storage_impl
         self.TYPE = self._storage_impl.TYPE
         self._ca_store = flow_datastore.ca_store