Fixing bug fixes along with a few enhancements (#56)

Author: vijayvammi

docs/RELEASES.md

@@ -1,5 +1,11 @@
 # Release history
 
+## v0.3.9 (2022-06-29)
+
+- Bug fix in catalog get function, Issue 54.
+- Enhancement of sending log levels to different executing environments, Issue 53
+- Map variable sent in as environment variable to non-shell tasks, Issue 51
+
 
 ## v0.3.8 (2022-06-29)
 

magnus/catalog.py

@@ -216,7 +216,7 @@ def get(self, name: str, run_id: str, compute_data_folder=None, **kwargs) -> Lis
             raise Exception(msg)
 
         catalog_location = self.get_catalog_location()
-        run_catalog = Path(catalog_location) / run_id
+        run_catalog = Path(catalog_location) / run_id / copy_to
 
         logger.debug(f'Copying objects to {copy_to} from the run catalog location of {run_catalog}')
 

magnus/tasks.py

@@ -88,6 +88,9 @@ def execute_command(self, map_variable: dict = None, **kwargs):
         parameters = self.get_parameters()
         filtered_parameters = utils.filter_arguments_for_func(f, parameters, map_variable)
 
+        if map_variable:
+            os.environ[defaults.PARAMETER_PREFIX + 'MAP_VARIABLE'] = json.dumps(map_variable)
+
         logger.info(f'Calling {func} from {module} with {filtered_parameters}')
         try:
             user_set_parameters = f(**filtered_parameters)
@@ -99,6 +102,9 @@ def execute_command(self, map_variable: dict = None, **kwargs):
             logger.exception(_e)
             raise
 
+        if map_variable:
+            del os.environ[defaults.PARAMETER_PREFIX + 'MAP_VARIABLE']
+
         self.set_parameters(user_set_parameters)
 
 
@@ -121,6 +127,9 @@ def execute_command(self, map_variable: dict = None, **kwargs):
         parameters = self.get_parameters()
         filtered_parameters = utils.filter_arguments_for_func(f, parameters, map_variable)
 
+        if map_variable:
+            os.environ[defaults.PARAMETER_PREFIX + 'MAP_VARIABLE'] = json.dumps(map_variable)
+
         logger.info(f'Calling lambda function: {self.command} with {filtered_parameters}')
         try:
             user_set_parameters = f(**filtered_parameters)
@@ -132,6 +141,9 @@ def execute_command(self, map_variable: dict = None, **kwargs):
             logger.exception(_e)
             raise
 
+        if map_variable:
+            del os.environ[defaults.PARAMETER_PREFIX + 'MAP_VARIABLE']
+
         self.set_parameters(user_set_parameters)
 
 
@@ -158,6 +170,9 @@ def execute_command(self, map_variable: dict = None, **kwargs):
             filtered_parameters = utils.filter_arguments_from_parameters(parameters=parameters,
                                                                          signature_parameters=notebook_parameters,
                                                                          map_variable=map_variable)
+            if map_variable:
+                os.environ[defaults.PARAMETER_PREFIX + 'MAP_VARIABLE'] = json.dumps(map_variable)
+
             notebook_output_path = self.config.get('notebook_output_path',
                                                    ''.join(self.command.split('.')[:-1]) + '_out.ipynb')
             kernel = self.config.get('notebook_kernel', None)
@@ -175,6 +190,10 @@ def execute_command(self, map_variable: dict = None, **kwargs):
                 kwds['kernel_name'] = kernel
 
             pm.execute_notebook(**kwds)
+
+            if map_variable:
+                del os.environ[defaults.PARAMETER_PREFIX + 'MAP_VARIABLE']
+
         except ImportError as e:
             msg = (
                 f'Task type of notebook requires papermill to be installed. Please install via optional: notebook'
@@ -194,9 +213,12 @@ def execute_command(self, map_variable: dict = None, **kwargs):
         # It might be that we have to write a bash/windows script that does things for us
         # Need to over-ride set parameters too
         subprocess_env = os.environ.copy()
+
         if map_variable:
             subprocess_env[defaults.PARAMETER_PREFIX + 'MAP_VARIABLE'] = json.dumps(map_variable)
+
         result = subprocess.run(self.command, check=True, env=subprocess_env, shell=True,
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
+
         print(result.stdout)
         print(result.stderr)

magnus/utils.py

@@ -442,8 +442,11 @@ def get_node_execution_command(executor, node, map_variable=None, over_write_run
     if over_write_run_id:
         run_id = over_write_run_id
 
+    log_level = logging.getLevelName(logger.getEffectiveLevel())
+
     action = (f'magnus execute_single_node {run_id} '
-              f'{node.command_friendly_name()} --file {executor.pipeline_file}'
+              f'{node.command_friendly_name()} --file {executor.pipeline_file} '
+              f'--log-level {log_level}'
               )
 
     if executor.variables_file:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "magnus"
-version = "0.3.8"
+version = "0.3.9"
 description = "A Compute agnostic pipelining software"
 authors = ["Vijay Vammi <vijay.vammi@astrazeneca.com>", "Vijay Vammi <mesanthu@gmail.com>"]
 license = "Apache-2.0"