From 12cf734ec47c3025dc175a59ffde02c92e185248 Mon Sep 17 00:00:00 2001 From: benpankow Date: Tue, 28 May 2024 15:00:25 -0700 Subject: [PATCH] break up default building --- .../pythonic_config/conversion_utils.py | 79 +++++++++++-------- 1 file changed, 48 insertions(+), 31 deletions(-) diff --git a/python_modules/dagster/dagster/_config/pythonic_config/conversion_utils.py b/python_modules/dagster/dagster/_config/pythonic_config/conversion_utils.py index e72ccfbabac96..e71f2ec3f8a39 100644 --- a/python_modules/dagster/dagster/_config/pythonic_config/conversion_utils.py +++ b/python_modules/dagster/dagster/_config/pythonic_config/conversion_utils.py @@ -34,48 +34,65 @@ from dagster._utils.typing_api import is_closed_python_optional_type +def _create_new_default_from_subfields( + old_field: Field, updated_sub_fields: dict[str, Field], additional_default_values: dict +) -> Any: + """Generates a replacement default value for a field based on its updated subfields. + + If all required subfields have defaults, or the field is a Selector, + then we can construct a new default value composed of the defaults of the subfields. + """ + new_default = ( + old_field.default_value if old_field.default_provided else FIELD_NO_DEFAULT_PROVIDED + ) + + if all( + sub_field.default_provided or not sub_field.is_required + for sub_field in updated_sub_fields.values() + ) or ( + old_field.config_type.kind == ConfigTypeKind.SELECTOR + and any(sub_field.default_provided for sub_field in updated_sub_fields.values()) + ): + new_default = { + **additional_default_values, + **{k: v.default_value for k, v in updated_sub_fields.items() if v.default_provided}, + } + + return new_default + + # This is from https://github.com/dagster-io/dagster/pull/11470 def _apply_defaults_to_schema_field(old_field: Field, additional_default_values: Any) -> Field: """Given a config Field and a set of default values (usually a dictionary or raw default value), return a new Field with the default values applied to it (and recursively to any sub-fields). """ - # If the field has subfields and the default value is a dictionary, iterate - # over the subfields and apply the defaults to them. - if isinstance(old_field.config_type, _ConfigHasFields) and isinstance( + # Any config type which does not have subfields or which doesn't have a new default value + # to apply can be copied directly. + if not isinstance(old_field.config_type, _ConfigHasFields) or not isinstance( additional_default_values, dict ): - updated_sub_fields = { - k: _apply_defaults_to_schema_field( - sub_field, additional_default_values.get(k, FIELD_NO_DEFAULT_PROVIDED) - ) - for k, sub_field in old_field.config_type.fields.items() - } + return copy_with_default(old_field, additional_default_values) - # We also apply a new default value to the field if all of its subfields have defaults - new_default = ( - old_field.default_value if old_field.default_provided else FIELD_NO_DEFAULT_PROVIDED + # If the field has subfields and the default value is a dictionary, iterate + # over the subfields and apply the defaults to them. + updated_sub_fields = { + k: _apply_defaults_to_schema_field( + sub_field, additional_default_values.get(k, FIELD_NO_DEFAULT_PROVIDED) ) + for k, sub_field in old_field.config_type.fields.items() + } - if all( - sub_field.default_provided or not sub_field.is_required - for sub_field in updated_sub_fields.values() - ) or ( - old_field.config_type.kind == ConfigTypeKind.SELECTOR - and any(sub_field.default_provided for sub_field in updated_sub_fields.values()) - ): - new_default = { - **additional_default_values, - **{k: v.default_value for k, v in updated_sub_fields.items() if v.default_provided}, - } + # We also compute a replacement default value, if possible. + new_default_value = _create_new_default_from_subfields( + old_field, updated_sub_fields, additional_default_values + ) - return Field( - config=old_field.config_type.__class__(fields=updated_sub_fields), - default_value=new_default, - is_required=new_default == FIELD_NO_DEFAULT_PROVIDED and old_field.is_required, - description=old_field.description, - ) - else: - return copy_with_default(old_field, additional_default_values) + return Field( + config=old_field.config_type.__class__(fields=updated_sub_fields), + default_value=new_default_value, + is_required=new_default_value == FIELD_NO_DEFAULT_PROVIDED and old_field.is_required, + description=old_field.description, + ) def copy_with_default(old_field: Field, new_config_value: Any) -> Field: