feat(config): add env block to deployment config

bentoctl/cli/__init__.py

@@ -219,3 +219,6 @@ def apply(deployment_config_file):
 
 # subcommands
 bentoctl.add_command(get_operator_management_subcommands())
+
+if __name__ == "__main__":
+    bentoctl()

bentoctl/cli/interactive.py

@@ -168,15 +168,34 @@ def prompt_input(
         input_value = {}
         if not belongs_to_list:
             indented_print(f"{field_name}:", indent_level)
-            for key in rule.get("schema").keys():
-                input_value[key] = prompt_input(
-                    key, rule.get("schema").get(key), indent_level + 1
-                )
+            if "schema" in rule:
+                for key in rule["schema"].keys():
+                    input_value[key] = prompt_input(
+                        key, rule["schema"].get(key, {}), indent_level + 1
+                    )
+            else:
+                # We don't have a schema so this is a dict with arbitrary keys.
+                help_message = rule.get("help_message")
+                with display_console_message(PromptMsg(help_message, None)):
+                    should_add_item_to_dict = prompt_confirmation(
+                        f"Do you want to add item to {field_name}"
+                    )
+                while should_add_item_to_dict:
+                    key = prompt_input_value("key", rule.get("keysrules", {}))
+                    value = prompt_input_value("value", rule.get("valuesrules", {}))
+                    display_input_result = f"{key}: {value}"
+                    indented_print(display_input_result, indent_level=indent_level + 1)
+                    input_value[key] = value
+
+                    with display_console_message(PromptMsg(help_message, None)):
+                        should_add_item_to_dict = prompt_confirmation(
+                            f"Do you want to add item to {field_name}"
+                        )
         else:
-            for i, key in enumerate(rule.get("schema").keys()):
+            for i, key in enumerate(rule["schema"].keys()):
                 input_value[key] = prompt_input(
                     key,
-                    rule.get("schema").get(key),
+                    rule["schema"].get(key, {}),
                     indent_level,
                     belongs_to_list,
                     i == 0,  # require display '-' for first item of a dict
@@ -276,4 +295,7 @@ def deployment_config_builder():
     spec = generate_spec(operator.schema)
     deployment_config["spec"] = dict(spec)
 
+    env = prompt_input("[b]env", deployment_config_schema.get("env"), indent_level=0)
+    deployment_config["env"] = dict(env)
+
     return DeploymentConfig(deployment_config)

bentoctl/deployment_config.py

@@ -54,6 +54,21 @@ def operator_exists(field, operator_name, error):
     "spec": {
         "required": True,
     },
+    "env": {
+        "required": False,
+        "nullable": True,
+        "type": "dict",
+        "keysrules": {
+            "type": "string",
+            "regex": "[a-zA-Z_]{1,}[a-zA-Z0-9_]{0,}",
+            "help_message": "An environment variable name to set for deployment",
+        },
+        "valuesrules": {
+            "type": "string",
+            "help_message": "An environment variable value to set for deployment",
+        },
+        "help_message": "Used to set runtime environment variables for deployment",
+    },
 }
 
 
@@ -62,14 +77,19 @@ def remove_help_message(schema):
     Remove the help_messages in the validation dict.
     """
     for field, rules in schema.items():
+        if not isinstance(rules, dict):
+            continue
+
         if "help_message" in rules:
             del rules["help_message"]
         if rules.get("type") == "dict":
-            rules["schema"] = remove_help_message(rules.get("schema"))
+            for key in ("schema", "keysrules", "valuesrules"):
+                if key in rules:
+                    rules[key] = remove_help_message(rules[key])
         elif rules.get("type") == "list":
-            rules["schema"] = remove_help_message({"list_item": rules.get("schema")})[
-                "list_item"
-            ]
+            rules["schema"] = remove_help_message(
+                {"list_item": rules.get("schema", {})}
+            )["list_item"]
         schema[field] = rules
     return schema
 
@@ -103,6 +123,7 @@ def __init__(self, deployment_config: t.Dict[str, t.Any]):
         self._set_name()
         self._set_operator()
         self._set_template_type()
+        self._set_env()
         self._set_operator_spec()
 
     def _set_name(self):
@@ -143,6 +164,19 @@ def _set_template_type(self):
                 )
             )
 
+    def _set_env(self):
+        copied_env = copy.deepcopy(self.deployment_config.get("env"))
+        validated_env = None
+        if copied_env is not None:
+            v = cerberus.Validator()
+            env_schema = remove_help_message(
+                {"env": copy.deepcopy(deployment_config_schema["env"])}
+            )
+            validated_env = v.validated({"env": copied_env}, env_schema)["env"]
+            if validated_env is None:
+                raise InvalidDeploymentConfig(config_errors=v.errors)
+        self.env = validated_env
+
     def _set_operator_spec(self):
         # cleanup operator_schema by removing 'help_message' field
         operator_schema = remove_help_message(schema=self.operator.schema)
@@ -151,6 +185,13 @@ def _set_operator_spec(self):
         validated_spec = v.validated(copied_operator_spec, schema=operator_schema)
         if validated_spec is None:
             raise InvalidDeploymentConfig(config_errors=v.errors)
+
+        # We add `env` through the operator spec to avoid introducing an
+        # additional argument to every operator which would be a breaking change.
+        # TODO: introduce the breaking change to clean up the interface.
+        if self.env is not None:
+            validated_spec["env"] = self.env
+
         self.operator_spec = validated_spec
 
     def set_bento(self, bento_tag: str):

tests/unit/cli/test_interactive.py

@@ -44,7 +44,7 @@ def test_deployment_config_builder(
 ):
     operator1 = mock_operator(name="operator1", schema={})
     get_mock_operator_registry.get = lambda _: operator1
-    stdin = StringIO("testdeployment\n")
+    stdin = StringIO("testdeployment\ny\nBENTOML_API_WORKERS\n1\nn\n")
     monkeypatch.setattr("sys.stdin", stdin)
     monkeypatch.setattr(
         interactive_cli, "dropdown_select", lambda field, options: "operator1"
@@ -97,6 +97,19 @@ def test_deployment_config_builder(
     },
 }
 
+single_map = {
+    "rule": {
+        "type": "dict",
+        "keysrules": {"type": "string"},
+        "valuesrules": {"type": "integer", "coerce": int},
+    },
+    "user_input": "y\nkey_1\n1\ny\nkey_2\n2\nn",
+    "expected_output": {
+        "key_1": 1,
+        "key_2": 2,
+    },
+}
+
 all_values = {
     "rule": {
         "type": "dict",
@@ -118,6 +131,7 @@ def test_deployment_config_builder(
         list_of_strings_required,
         list_of_strings_required_with_nested_dict,
         nested_dict,
+        single_map,
         all_values,
     ],
 )