Add PermissiveDict type to permit config with unknown parameters (#1024)

python_modules/dagster/dagster/__init__.py

@@ -63,6 +63,7 @@
     output_schema,
     output_selector_schema,
     Path,
+    PermissiveDict,
     PythonObjectType,
     Selector,
     String,
@@ -131,6 +132,7 @@
     'output_schema',
     'output_selector_schema',
     'Path',
+    'PermissiveDict',
     'PythonObjectType',
     'Selector',
     'String',

python_modules/dagster/dagster/core/types/__init__.py

@@ -6,7 +6,7 @@
     output_selector_schema,
 )
 from .field import Field
-from .field_utils import Dict, NamedDict, Selector, NamedSelector
+from .field_utils import Dict, NamedDict, Selector, NamedSelector, PermissiveDict
 from .runtime import PythonObjectType
 from .wrapping import Nullable, List
 

python_modules/dagster/dagster/core/types/default_applier.py

@@ -63,6 +63,14 @@ def apply_defaults_to_composite_type(composite_type, config_value):
         elif not field_def.is_optional:
             check.failed('Missing non-optional composite member not caught in validation')
 
+    # For permissive composite fields, we skip applying defaults because these fields are unknown
+    # to us
+    if composite_type.is_permissive_composite:
+        defined_fields = set(fields.keys())
+        extra_fields = incoming_fields - defined_fields
+        for extra_field in extra_fields:
+            processed_fields[extra_field] = config_value[extra_field]
+
     return processed_fields
 
 

python_modules/dagster/dagster/core/types/evaluator.py

@@ -454,10 +454,16 @@ def validate_composite_config_value(composite_type, config_value, stack):
     defined_fields = set(fields.keys())
     incoming_fields = set(config_value.keys())
 
-    for received_field in incoming_fields:
-        if received_field not in defined_fields:
-            yield create_field_not_defined_error(composite_type, stack, received_field)
-
+    # Here, we support permissive composites. In cases where we know the set of permissible keys a
+    # priori, we validate against the config. For permissive composites, we give the user an escape
+    # hatch where they can specify arbitrary fields...
+    if not composite_type.is_permissive_composite:
+        for received_field in incoming_fields:
+            if received_field not in defined_fields:
+                yield create_field_not_defined_error(composite_type, stack, received_field)
+
+    # ...However, for any fields the user *has* told us about, we validate against their config
+    # specification.
     for expected_field, field_def in fields.items():
         if expected_field in incoming_fields:
             for error in _validate_config(

python_modules/dagster/dagster/core/types/field_utils.py

@@ -131,6 +131,10 @@ class _ConfigComposite(_ConfigHasFields):
     def is_composite(self):
         return True
 
+    @property
+    def is_permissive_composite(self):
+        return False
+
 
 class _ConfigSelector(_ConfigHasFields):
     @property
@@ -187,6 +191,30 @@ def __init__(self):
     return _Dict
 
 
+def PermissiveDict(fields=None):
+    '''A permissive dict will permit the user to partially specify the permitted fields. Any fields
+    that are specified and passed in will be type checked. Other fields will be allowed, but
+    will be ignored by the type checker.
+    '''
+
+    class _PermissiveDict(_ConfigComposite):
+        def __init__(self):
+            key = 'PermissiveDict.' + str(DictCounter.get_next_count())
+            super(_PermissiveDict, self).__init__(
+                name=None,
+                key=key,
+                fields=fields or dict(),
+                description='A configuration dictionary with typed fields',
+                type_attributes=ConfigTypeAttributes(is_builtin=True),
+            )
+
+        @property
+        def is_permissive_composite(self):
+            return True
+
+    return _PermissiveDict
+
+
 def Selector(fields):
     '''Selectors are used when you want to be able present several different options to the user but
     force them to select one. For example, it would not make much sense to allow them

python_modules/dagster/dagster_tests/core_tests/types_tests/test_config_type_system.py

@@ -9,6 +9,7 @@
     Int,
     List,
     Nullable,
+    PermissiveDict,
     PipelineConfigEvaluationError,
     PipelineContextDefinition,
     PipelineDefinition,
@@ -88,6 +89,10 @@ def _mixed_required_optional_string_config_dict_with_default():
     )
 
 
+def _multiple_required_fields_config_permissive_dict():
+    return Field(PermissiveDict({'field_one': Field(String), 'field_two': Field(String)}))
+
+
 def _validate(config_field, value):
     return throwing_evaluate_config_value(config_field.config_type, value)
 
@@ -146,6 +151,12 @@ def test_multiple_required_fields_failing():
     with pytest.raises(DagsterEvaluateConfigValueError):
         _validate(_multiple_required_fields_config_dict(), {'field_one': 'yup', 'extra': 'yup'})
 
+    with pytest.raises(DagsterEvaluateConfigValueError):
+        _validate(
+            _multiple_required_fields_config_dict(),
+            {'field_one': 'yup', 'field_two': 'yup', 'extra': 'should_not_exist'},
+        )
+
     with pytest.raises(DagsterEvaluateConfigValueError):
         _validate(
             _multiple_required_fields_config_dict(), {'field_one': 'value_one', 'field_two': 2}
@@ -183,6 +194,50 @@ def test_single_optional_field_passing_with_default():
     ) == {'optional_field': 'override'}
 
 
+def test_permissive_multiple_required_fields_passing():
+    assert _validate(
+        _multiple_required_fields_config_permissive_dict(),
+        {
+            'field_one': 'value_one',
+            'field_two': 'value_two',
+            'previously_unspecified': 'should_exist',
+        },
+    ) == {
+        'field_one': 'value_one',
+        'field_two': 'value_two',
+        'previously_unspecified': 'should_exist',
+    }
+
+
+def test_permissive_multiple_required_fields_nested_passing():
+    assert _validate(
+        _multiple_required_fields_config_permissive_dict(),
+        {
+            'field_one': 'value_one',
+            'field_two': 'value_two',
+            'previously_unspecified': {'nested': 'value', 'with_int': 2},
+        },
+    ) == {
+        'field_one': 'value_one',
+        'field_two': 'value_two',
+        'previously_unspecified': {'nested': 'value', 'with_int': 2},
+    }
+
+
+def test_permissive_multiple_required_fields_failing():
+    with pytest.raises(DagsterEvaluateConfigValueError):
+        _validate(_multiple_required_fields_config_permissive_dict(), {})
+
+    with pytest.raises(DagsterEvaluateConfigValueError):
+        _validate(_multiple_required_fields_config_permissive_dict(), {'field_one': 'yup'})
+
+    with pytest.raises(DagsterEvaluateConfigValueError):
+        _validate(
+            _multiple_required_fields_config_permissive_dict(),
+            {'field_one': 'value_one', 'field_two': 2},
+        )
+
+
 def test_mixed_args_passing():
     assert _validate(
         _mixed_required_optional_string_config_dict_with_default(),