fix: string format in configuration

factorpricingmodel · Dec 14, 2022 · bd952c8 · bd952c8
1 parent 272e6ac
commit bd952c8
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 2 deletions.
diff --git a/src/prefect_yaml/flow/__init__.py b/src/prefect_yaml/flow/__init__.py
@@ -69,7 +69,13 @@ def is_args(kwargs):
         return output_obj.load()
 
     # Run the task to prepare the value
-    caller = description["caller"]
+    try:
+        caller = description["caller"]
+    except KeyError:
+        raise RuntimeError(
+            f"Task {name} does not have any caller but it does not "
+            f"exist in the path {output_obj.output_path}."
+        )
     module_name, function_name = caller.split(":")
     module = import_module(module_name)
     func = getattr(module, function_name)

diff --git a/src/prefect_yaml/flow/config.py b/src/prefect_yaml/flow/config.py
@@ -95,6 +95,7 @@ def load_configuration(
         yaml = YAML()
         yaml.register_class(Data)
         configuration = yaml.load(configuration)
+        configuration = format_string(configuration, variables)
         # Create the data object if it hasn't been depended
         if "metadata" not in configuration:
             raise ValueError("Key metadata not found in configuration")
@@ -104,7 +105,6 @@ def load_configuration(
             Data.create(task_name)
         # Update the descriptions of all the data objects
         configuration = update_data(configuration)
-        configuration = format_string(configuration, variables)
         data_cache = _DATA_CACHE
     finally:
         _DATA_CACHE = {}

diff --git a/tests/test_config.py b/tests/test_config.py
@@ -82,6 +82,8 @@ def test_load_configuration_variables():
             directory: "{output_directory}"
     task:
       task_d:
+        output:
+          name: "{name}"
         caller: itertools:chain
         parameters:
           - "{param1}"
@@ -93,7 +95,9 @@ def test_load_configuration_variables():
             "output_directory": ".output",
             "param1": "ABC",
             "param2": "DEF",
+            "name": "task_a",
         },
     )
     assert configuration["metadata"]["output"]["directory"] == ".output"
     assert configuration["task"]["task_d"]["parameters"] == ["ABC", "DEF"]
+    assert configuration["task"]["task_d"]["output"]["name"] == "task_a"