[secret-resource-param] core

Author: smackesey

python_modules/dagster/dagster/_config/field.py

@@ -237,6 +237,9 @@ class Field:
         description (str):
             A human-readable description of this config field.
 
+        is_secret (bool):
+            Whether this field contains sensitive data that should be masked in UIs. Defaults to False.
+
     Examples:
         .. code-block:: python
 
@@ -268,13 +271,15 @@ def __init__(
         default_value: Any = FIELD_NO_DEFAULT_PROVIDED,
         is_required: Optional[bool] = None,
         description: Optional[str] = None,
+        is_secret: bool = False,
     ):
         from dagster._config.post_process import resolve_defaults
         from dagster._config.validate import validate_config
 
         self.config_type = check.inst(self._resolve_config_arg(config), ConfigType)
 
         self._description = check.opt_str_param(description, "description")
+        self._is_secret = check.bool_param(is_secret, "is_secret")
 
         check.opt_bool_param(is_required, "is_required")
 
@@ -367,6 +372,12 @@ def description(self) -> Optional[str]:
         """A human-readable description of this config field, if provided."""
         return self._description
 
+    @public
+    @property
+    def is_secret(self) -> bool:
+        """Whether this field contains sensitive data that should be masked in UIs."""
+        return self._is_secret
+
     @property
     def default_value_as_json_str(self) -> str:
         check.invariant(self.default_provided, "Asking for default value when none was provided")

python_modules/dagster/dagster/_config/pythonic_config/conversion_utils.py

@@ -144,13 +144,18 @@ def _convert_pydantic_field(
         if isinstance(default_to_pass, Enum):
             default_to_pass = default_to_pass.name
 
+        # Extract is_secret from json_schema_extra
+        extras = pydantic_field.json_schema_extra or {}
+        is_secret = bool(extras.get("is_secret", False))
+
         return Field(
             config=config_type,
             description=pydantic_field.description,
             is_required=pydantic_field.is_required()
             and not is_closed_python_optional_type(field_type)
             and default_to_pass == FIELD_NO_DEFAULT_PROVIDED,
             default_value=default_to_pass,
+            is_secret=is_secret,
         )
 
 

python_modules/dagster/dagster/_config/snap.py

@@ -169,7 +169,7 @@ class ConfigEnumValueSnap:
     description: Optional[str]
 
 
-@whitelist_for_serdes
+@whitelist_for_serdes(skip_when_none_fields={"is_secret"})
 @record
 class ConfigFieldSnap:
     name: Optional[str]
@@ -178,6 +178,7 @@ class ConfigFieldSnap:
     default_provided: bool
     default_value_as_json_str: Optional[str]
     description: Optional[str]
+    is_secret: Optional[bool] = None
 
 
 def snap_from_field(name: str, field: Field):
@@ -190,6 +191,7 @@ def snap_from_field(name: str, field: Field):
             field.default_value_as_json_str if field.default_provided else None
         ),
         description=field.description,
+        is_secret=field.is_secret if field.is_secret else None,
     )
 
 

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

@@ -1188,3 +1188,14 @@ def test_order(context):
     assert wrap_op_in_graph_and_execute(
         test_order, run_config={"ops": {"test_order": {"config": alphabet}}}
     ).success
+
+
+def test_field_is_secret_parameter():
+    """Test that Field accepts and stores is_secret parameter."""
+    # Non-secret field
+    regular_field = dg.Field(str, description="A regular field")
+    assert regular_field.is_secret is False
+
+    # Secret field
+    secret_field = dg.Field(str, description="A secret field", is_secret=True)
+    assert secret_field.is_secret is True

python_modules/dagster/dagster_tests/core_tests/resource_tests/pythonic_resources/test_general_pythonic_resources.py

@@ -13,7 +13,10 @@
     InitResourceContext,
 )
 from dagster._check import CheckError
-from dagster._config.pythonic_config import ConfigurableResourceFactory
+from dagster._config.pythonic_config import (
+    ConfigurableResourceFactory,
+    infer_schema_from_config_class,
+)
 from dagster._utils.cached_method import cached_method
 from pydantic import (
     Field as PyField,
@@ -1119,3 +1122,33 @@ class MergeResource(dg.ConfigurableResource):
         int_res=StrResource.configure_at_launch(),
         str_res=IntResource.configure_at_launch(),
     )
+
+
+def test_secret_field():
+    """Test that is_secret is extracted from json_schema_extra in ConfigurableResource."""
+
+    class MyResource(dg.ConfigurableResource):
+        api_key: str = PyField(
+            description="API key for authentication",
+            json_schema_extra={"is_secret": True},
+        )
+        host: str = PyField(description="Host URL")
+        password: str = PyField(json_schema_extra={"is_secret": True})
+
+    # Get the inferred schema
+    schema_field = infer_schema_from_config_class(MyResource)
+
+    # The schema should have the fields
+    fields_dict = schema_field.config_type.fields  # type: ignore
+
+    # Check that api_key is marked as secret
+    assert "api_key" in fields_dict
+    assert fields_dict["api_key"].is_secret is True
+
+    # Check that host is not marked as secret
+    assert "host" in fields_dict
+    assert fields_dict["host"].is_secret is False
+
+    # Check that password is marked as secret
+    assert "password" in fields_dict
+    assert fields_dict["password"].is_secret is True

python_modules/dagster/dagster_tests/core_tests/snap_tests/test_config_schema_snapshot.py

@@ -67,10 +67,11 @@ def test_field_things():
             "opt": dg.Field(int, is_required=False),
             "opt_with_default": dg.Field(int, is_required=False, default_value=2),
             "req_with_desc": dg.Field(int, description="A desc"),
+            "secret_field": dg.Field(str, is_secret=True),
         }
     )
 
-    assert dict_snap.fields and len(dict_snap.fields) == 4
+    assert dict_snap.fields and len(dict_snap.fields) == 5
 
     field_snap_dict = {field_snap.name: field_snap for field_snap in dict_snap.fields}
 
@@ -85,6 +86,8 @@ def test_field_things():
 
     assert field_snap_dict["req_with_desc"].is_required is True
     assert field_snap_dict["req_with_desc"].description == "A desc"
+    assert field_snap_dict["secret_field"].is_secret is True
+    assert not field_snap_dict["req"].is_secret
 
 
 def test_basic_list():

python_modules/libraries/dagster-shared/dagster_shared/dagster_model/pydantic_compat_layer.py

@@ -60,6 +60,10 @@ def discriminator(self) -> Optional[str]:
         if hasattr(self.field, "discriminator"):
             return self.field.discriminator if hasattr(self.field, "discriminator") else None
 
+    @property
+    def json_schema_extra(self) -> Optional[dict[str, Any]]:
+        return getattr(self.field, "json_schema_extra", None)
+
 
 def model_fields(model: type[BaseModel]) -> dict[str, ModelFieldCompat]:
     """Returns a dictionary of fields for a given pydantic model, wrapped