avoid race condition with argo-workflows

metaflow/plugins/pypi/conda_decorator.py

@@ -122,7 +122,7 @@ def step_init(self, flow, graph, step, decos, environment, flow_datastore, logge
 
         environment.set_local_root(LocalStorage.get_datastore_root_from_config(logger))
 
-        self.disabled = not self.environment.get_environment(
+        self.disabled = self.environment.is_disabled(
             next(step for step in self.flow if step.name == self.step)
         )
 

metaflow/plugins/pypi/conda_environment.py

@@ -181,6 +181,15 @@ def interpreter(self, step_name):
         # User workloads are executed through the conda environment's interpreter.
         return self.solvers["conda"].interpreter(id_)
 
+    def is_disabled(self, step):
+        for decorator in step.decorators:
+            # @conda decorator is guaranteed to exist thanks to self.decospecs
+            if decorator.name in ["conda", "pypi"]:
+                # handle @conda/@pypi(disabled=True)
+                disabled = decorator.attributes["disabled"]
+                return disabled or str(disabled).lower() != "false"
+        return False
+
     @functools.lru_cache(maxsize=None)
     def get_environment(self, step):
         environment = {}