mage-ai · wangxiaoyou1993 · Jul 15, 2022 · Jul 14, 2022 · Jul 15, 2022 · Jul 15, 2022
diff --git a/mage_ai/command_line.py b/mage_ai/command_line.py
@@ -37,7 +37,7 @@ def main():
         sys.path.append(os.path.dirname(project_path))
         pipeline = Pipeline(pipeline_uuid, project_path)
 
-        asyncio.run(pipeline.execute(analyze_outputs=False))
+        asyncio.run(pipeline.execute(analyze_outputs=False, update_status=False))
 
 
 if __name__ == "__main__":

diff --git a/mage_ai/data_preparation/models/block.py b/mage_ai/data_preparation/models/block.py
@@ -157,29 +157,45 @@ def delete(self):
             p.delete_block(p.get_block(self.uuid))
         os.remove(self.file_path)
 
-    def execute_sync(self, analyze_outputs=True, custom_code=None, redirect_outputs=False):
+    def execute_sync(
+        self,
+        analyze_outputs=True,
+        custom_code=None,
+        redirect_outputs=False,
+        update_status=True,
+    ):
         try:
             output = self.execute_block(custom_code=custom_code, redirect_outputs=redirect_outputs)
             block_output = output['output']
             self.__verify_outputs(block_output)
             variable_mapping = dict(zip(self.output_variables.keys(), block_output))
             self.__store_variables(variable_mapping)
-            self.status = BlockStatus.EXECUTED
+            if update_status:
+                self.status = BlockStatus.EXECUTED
             if analyze_outputs:
                 self.__analyze_outputs(variable_mapping)
         except Exception as err:
-            self.status = BlockStatus.FAILED
+            if update_status:
+                self.status = BlockStatus.FAILED
             raise Exception(f'Exception encountered in block {self.uuid}') from err
         finally:
-            self.__update_pipeline_block()
+            if update_status:
+                self.__update_pipeline_block()
         return output
 
-    async def execute(self, analyze_outputs=True, custom_code=None, redirect_outputs=False):
+    async def execute(
+        self,
+        analyze_outputs=True,
+        custom_code=None,
+        redirect_outputs=False,
+        update_status=True,
+    ):
         with VerboseFunctionExec(f'Executing {self.type} block: {self.uuid}'):
             return self.execute_sync(
                 analyze_outputs=analyze_outputs,
                 custom_code=custom_code,
                 redirect_outputs=redirect_outputs,
+                update_status=update_status,
             )
 
     def __validate_execution(self, decorated_functions, input_vars):

diff --git a/mage_ai/data_preparation/models/pipeline.py b/mage_ai/data_preparation/models/pipeline.py
@@ -92,7 +92,13 @@ def block_deletable(self, block):
             return True
         return len(self.blocks_by_uuid[block.uuid].downstream_blocks) == 0
 
-    async def execute(self, analyze_outputs=True, run_all_blocks=False, redirect_outputs=False):
+    async def execute(
+        self,
+        analyze_outputs=True,
+        run_all_blocks=False,
+        redirect_outputs=False,
+        update_status=True,
+    ):
         """
         Async function for parallel processing
         This function will schedule the block execution in topological
@@ -118,7 +124,11 @@ async def execute(self, analyze_outputs=True, run_all_blocks=False, redirect_out
                 continue
             await asyncio.gather(*[tasks[u.uuid] for u in block.upstream_blocks])
             task = asyncio.create_task(
-                block.execute(analyze_outputs=analyze_outputs, redirect_outputs=redirect_outputs)
+                block.execute(
+                    analyze_outputs=analyze_outputs,
+                    redirect_outputs=redirect_outputs,
+                    update_status=update_status,
+                )
             )
             tasks[block.uuid] = task
             for downstream_block in block.downstream_blocks:

diff --git a/mage_ai/data_preparation/templates/repo/metadata.yaml b/mage_ai/data_preparation/templates/repo/metadata.yaml
@@ -0,0 +1 @@
+variables_dir: ./
diff --git a/mage_ai/data_preparation/variable_manager.py b/mage_ai/data_preparation/variable_manager.py
@@ -3,11 +3,23 @@
 from typing import Any, Dict, List
 import os
 import pandas as pd
+import yaml
 
 
 class VariableManager:
     def __init__(self, repo_path=None):
         self.repo_path = repo_path or get_repo_path()
+        self.variables_dir = self.repo_path
+        try:
+            with open(os.path.join(self.repo_path, 'metadata.yaml')) as f:
+                repo_config = yaml.full_load(f) or {}
+                variables_dir = repo_config.get('variables_dir')
+                if variables_dir is not None:
+                    self.variables_dir = os.path.abspath(
+                        os.path.join(self.repo_path, variables_dir),
+                    )
+        except Exception:
+            pass
         # TODO: implement caching logic
 
     def add_variable(
@@ -87,7 +99,10 @@ def get_variables_by_block(self, pipeline_uuid: str, block_uuid: str) -> Dict[st
         return sorted([v.split('.')[0] for v in variables])
 
     def __pipeline_path(self, pipeline_uuid: str) -> str:
-        return os.path.join(self.repo_path, 'pipelines', pipeline_uuid)
+        path = os.path.join(self.variables_dir, 'pipelines', pipeline_uuid)
+        if not os.path.exists(path):
+            os.makedirs(path, exist_ok=True)
+        return path
 
 
 def get_global_variable(pipeline_uuid: str, key: str) -> Any:

diff --git a/requirements.txt b/requirements.txt
@@ -1,8 +1,8 @@
 Faker==4.14.0
-Flask==2.1.2
-Jinja2==3.1.2
+Flask~=1.1.2
+Jinja2~=3.0.3
 MarkupSafe==2.1.1
-Werkzeug==2.1.2
+Werkzeug~=2.0.0
 asyncio==3.4.3
 beautifulsoup4
 boto3==1.24.19
@@ -12,7 +12,7 @@ flask-cors==3.0.10
 importlib-metadata==4.11.3
 ipykernel==6.15.0
 ipython
-itsdangerous==2.1.2
+itsdangerous~=1.1.0
 joblib>=1.1.0
 jupyter_client==7.3.4
 google-cloud-bigquery==3.2.0