move assign_subschema_defaults to util.schema

st2common/st2common/util/config_loader.py

@@ -27,6 +27,7 @@
 from st2common.content import utils as content_utils
 from st2common.util import jinja as jinja_utils
 from st2common.util.schema import (
+    assign_subschema_default_values,
     get_flattened_array_items_schema,
     get_flattened_object_properties_schema,
 )
@@ -99,7 +100,7 @@ def _get_values_for_config(self, config_schema_db, config_db):
 
         # If config_schema is available we do a second pass and set default values for required
         # items which values are not provided / available in the config itself
-        config = self._assign_subschema_default_values(
+        config = assign_subschema_default_values(
             subschema=schema_values, instance=config
         )
         return config
@@ -185,97 +186,6 @@ def _assign_dynamic_config_values(self, schema, config, parent_keys=None):
 
         return config
 
-    def _assign_subschema_default_values(self, subschema, instance):
-        """
-        Assign default values for particular config if default values are provided in the config
-        schema and a value is not specified in the config.
-
-        Note: This method mutates instance argument in place.
-
-        :rtype: ``dict|list``
-        """
-        subschema_is_dict = isinstance(subschema, dict)
-        iterator = subschema.items() if subschema_is_dict else enumerate(subschema)
-
-        # _get_flattened_*_schema ensures that schema_item is always a dict
-        for schema_item_key, schema_item in iterator:
-            has_default_value = "default" in schema_item
-
-            if isinstance(instance, dict):
-                has_instance_value = schema_item_key in instance
-            else:
-                has_instance_value = schema_item_key < len(instance)
-
-            try:
-                instance_value = instance[schema_item_key]
-            except (KeyError, IndexError):
-                instance_value = None
-
-            schema_item_type = schema_item.get("type", None)
-            is_mutatable = schema_item_type in ["array", "object"]
-
-            instance_value = self._assign_default_values(
-                schema=schema_item,
-                instance=instance_value,
-                can_replace_instance=not has_instance_value,
-            )
-
-            if (
-                has_default_value
-                or has_instance_value
-                or (is_mutatable and instance_value is not None)
-            ):
-                instance[schema_item_key] = instance_value
-
-        return instance
-
-    def _assign_default_values(self, schema, instance, can_replace_instance=False):
-        has_default_value = "default" in schema
-        default_value = schema.get("default", None)
-        if has_default_value and can_replace_instance and instance is None:
-            # instance value is not provided, but default value is, use a default value
-            instance = default_value
-
-        schema_type = schema.get("type", None)
-
-        if schema_type == "object":
-            has_properties = schema.get("properties", None)
-            has_pattern_properties = schema.get("patternProperties", None)
-            has_additional_properties = schema.get("additionalProperties", None)
-
-            # Inspect nested object properties
-            if has_properties or has_pattern_properties or has_additional_properties:
-                if not instance:
-                    instance = {}
-
-                properties_schema = get_flattened_object_properties_schema(
-                    schema,
-                    object_keys=instance.keys(),
-                )
-
-                self._assign_subschema_default_values(
-                    subschema=properties_schema, instance=instance
-                )
-
-        elif schema_type == "array":
-            has_items = schema.get("items", None)
-            has_additional_items = schema.get("additionalItems", None)
-
-            # Inspect nested array items
-            if has_items or has_additional_items:
-                if not instance:
-                    instance = []
-
-                items_schema = get_flattened_array_items_schema(
-                    schema,
-                    items_count=len(instance),
-                )
-                self._assign_subschema_default_values(
-                    subschema=items_schema, instance=instance
-                )
-
-        return instance
-
     def _get_datastore_value_for_expression(self, key, value, config_schema_item=None):
         """
         Retrieve datastore value by first resolving the datastore expression and then retrieving

st2common/st2common/util/schema/__init__.py

@@ -45,6 +45,7 @@
     "validate",
     "get_flattened_array_items_schema",
     "get_flattened_object_properties_schema",
+    "assign_subschema_default_values",
 ]
 
 # https://github.com/json-schema/json-schema/blob/master/draft-04/schema
@@ -584,3 +585,94 @@ def get_flattened_array_items_schema(array_schema, items_count=0):
         )
 
     return flattened_items_schema
+
+
+def assign_subschema_default_values(subschema, instance):
+    """
+    Assign default values for particular config if default values are provided in the config
+    schema and a value is not specified in the config.
+
+    Note: This method mutates instance argument in place.
+
+    :rtype: ``dict|list``
+    """
+    subschema_is_dict = isinstance(subschema, dict)
+    iterator = subschema.items() if subschema_is_dict else enumerate(subschema)
+
+    # _get_flattened_*_schema ensures that schema_item is always a dict
+    for schema_item_key, schema_item in iterator:
+        has_default_value = "default" in schema_item
+
+        if isinstance(instance, dict):
+            has_instance_value = schema_item_key in instance
+        else:
+            has_instance_value = schema_item_key < len(instance)
+
+        try:
+            instance_value = instance[schema_item_key]
+        except (KeyError, IndexError):
+            instance_value = None
+
+        schema_item_type = schema_item.get("type", None)
+        is_mutatable = schema_item_type in ["array", "object"]
+
+        instance_value = _assign_default_values(
+            schema=schema_item,
+            instance=instance_value,
+            can_replace_instance=not has_instance_value,
+        )
+
+        if (
+            has_default_value
+            or has_instance_value
+            or (is_mutatable and instance_value is not None)
+        ):
+            instance[schema_item_key] = instance_value
+
+    return instance
+
+
+def _assign_default_values(schema, instance, can_replace_instance=False):
+    has_default_value = "default" in schema
+    default_value = schema.get("default", None)
+    if has_default_value and can_replace_instance and instance is None:
+        # instance value is not provided, but default value is, use a default value
+        instance = default_value
+
+    schema_type = schema.get("type", None)
+
+    if schema_type == "object":
+        has_properties = schema.get("properties", None)
+        has_pattern_properties = schema.get("patternProperties", None)
+        has_additional_properties = schema.get("additionalProperties", None)
+
+        # Inspect nested object properties
+        if has_properties or has_pattern_properties or has_additional_properties:
+            if not instance:
+                instance = {}
+
+            properties_schema = get_flattened_object_properties_schema(
+                schema,
+                object_keys=instance.keys(),
+            )
+
+            assign_subschema_default_values(
+                subschema=properties_schema, instance=instance
+            )
+
+    elif schema_type == "array":
+        has_items = schema.get("items", None)
+        has_additional_items = schema.get("additionalItems", None)
+
+        # Inspect nested array items
+        if has_items or has_additional_items:
+            if not instance:
+                instance = []
+
+            items_schema = get_flattened_array_items_schema(
+                schema,
+                items_count=len(instance),
+            )
+            assign_subschema_default_values(subschema=items_schema, instance=instance)
+
+    return instance