Hide non_argument_deps

python_modules/dagster/dagster/_annotations.py

@@ -70,8 +70,9 @@ def is_public(obj: Annotatable) -> bool:
 @dataclass
 class DeprecatedInfo:
     breaking_version: str
-    additional_warn_text: Optional[str] = None
-    subject: Optional[str] = None
+    hidden: bool
+    additional_warn_text: Optional[str]
+    subject: Optional[str]
 
 
 @overload
@@ -122,9 +123,14 @@ def deprecated(
             `emit_runtime_warning=False`, as we don't want to warn users when a
             deprecated API is used internally.
         emit_runtime_warning (bool): Whether to emit a warning when the function is called.
+        hidden (bool): Whether or not this is a hidden parameters. Hidden parameters are only
+            passed via kwargs and are hidden from the type signature. This makes it so
+            that this hidden parameter does not appear in typeaheads. In order to provide
+            high quality error messages we also provide the helper function
+            only_allow_hidden_params_in_kwargs to ensure there are high quality
+            error messages if the user passes an unsupported keyword argument.
 
     Usage:
-
         .. code-block:: python
 
             @deprecated(breaking_version="2.0", additional_warn_text="Use my_new_function instead")
@@ -140,6 +146,11 @@ def not_deprecated_function():
                 ...
                 some_deprecated_function()
                 ...
+
+            @deprecated_param(param="baz", breaking_version="2.0", hidden=True)
+            def func_with_hidden_args(**kwargs):
+                only_allow_hidden_params_in_kwargs(func_with_hidden_args, kwargs)
+                ...
     """
     if __obj is None:
         return lambda obj: deprecated(
@@ -154,7 +165,12 @@ def not_deprecated_function():
         setattr(
             target,
             _DEPRECATED_ATTR_NAME,
-            DeprecatedInfo(breaking_version, additional_warn_text, subject),
+            DeprecatedInfo(
+                breaking_version=breaking_version,
+                additional_warn_text=additional_warn_text,
+                subject=subject,
+                hidden=False,
+            ),
         )
 
         if emit_runtime_warning:
@@ -196,6 +212,7 @@ def deprecated_param(
     breaking_version: str,
     additional_warn_text: Optional[str] = ...,
     emit_runtime_warning: bool = ...,
+    hidden: bool = ...,
 ) -> T_Annotatable: ...
 
 
@@ -207,6 +224,7 @@ def deprecated_param(
     breaking_version: str,
     additional_warn_text: Optional[str] = ...,
     emit_runtime_warning: bool = ...,
+    hidden: bool = ...,
 ) -> Callable[[T_Annotatable], T_Annotatable]: ...
 
 
@@ -217,6 +235,7 @@ def deprecated_param(
     breaking_version: str,
     additional_warn_text: Optional[str] = None,
     emit_runtime_warning: bool = True,
+    hidden: bool = False,
 ) -> T_Annotatable:
     """Mark a parameter of a class initializer or function/method as deprecated. This appends some
     metadata to the decorated object that causes the specified argument to be rendered with a
@@ -233,6 +252,14 @@ def deprecated_param(
         additional_warn_text (str): Additional text to display after the deprecation warning.
             Typically this should suggest a newer API.
         emit_runtime_warning (bool): Whether to emit a warning when the function is called.
+        hidden (bool): Whether or not this is a hidden parameters. Hidden parameters are only
+            passed via kwargs and are hidden from the type signature. This makes it so
+            that this hidden parameter does not appear in typeaheads. In order to provide
+            high quality error messages we also provide the helper function
+            only_allow_hidden_params_in_kwargs to ensure there are high quality
+            error messages if the user passes an unsupported keyword argument.
+
+
     """
     if __obj is None:
         return lambda obj: deprecated_param(  # type: ignore
@@ -241,18 +268,22 @@ def deprecated_param(
             breaking_version=breaking_version,
             additional_warn_text=additional_warn_text,
             emit_runtime_warning=emit_runtime_warning,
+            hidden=hidden,
         )
     else:
-        check.invariant(
-            _annotatable_has_param(__obj, param),
-            f"Attempted to mark undefined parameter `{param}` deprecated.",
-        )
+        if not hidden:
+            check.invariant(
+                _annotatable_has_param(__obj, param),
+                f"Attempted to mark undefined parameter `{param}` deprecated.",
+            )
         target = _get_annotation_target(__obj)
         if not hasattr(target, _DEPRECATED_PARAM_ATTR_NAME):
             setattr(target, _DEPRECATED_PARAM_ATTR_NAME, {})
         getattr(target, _DEPRECATED_PARAM_ATTR_NAME)[param] = DeprecatedInfo(
             breaking_version=breaking_version,
             additional_warn_text=additional_warn_text,
+            hidden=hidden,
+            subject=None,
         )
 
         if emit_runtime_warning:
@@ -575,3 +606,17 @@ def _get_warning_stacklevel(obj: Annotatable):
 def _annotatable_has_param(obj: Annotatable, param: str) -> bool:
     target_fn = get_decorator_target(obj)
     return param in inspect.signature(target_fn).parameters
+
+
+def only_allow_hidden_params_in_kwargs(annotatable: Annotatable, kwargs: Mapping[str, Any]) -> None:
+    deprecated_params = (
+        get_deprecated_params(annotatable) if has_deprecated_params(annotatable) else {}
+    )
+    for param in kwargs:
+        if param not in deprecated_params:
+            raise TypeError(f"{annotatable.__name__} got an unexpected keyword argument '{param}'")
+
+        check.invariant(
+            deprecated_params[param].hidden,
+            f"Unexpected non-hidden deprecated parameter '{param}' in kwargs. Should never get here.",
+        )

python_modules/dagster/dagster/_core/definitions/decorators/asset_decorator.py

@@ -15,7 +15,11 @@
 )
 
 import dagster._check as check
-from dagster._annotations import deprecated_param, experimental_param
+from dagster._annotations import (
+    deprecated_param,
+    experimental_param,
+    only_allow_hidden_params_in_kwargs,
+)
 from dagster._config.config_schema import UserConfigSchema
 from dagster._core.definitions.asset_dep import AssetDep, CoercibleToAssetDep
 from dagster._core.definitions.auto_materialize_policy import AutoMaterializePolicy
@@ -55,6 +59,7 @@
 @overload
 def asset(
     compute_fn: Callable[..., Any],
+    **kwargs,
 ) -> AssetsDefinition: ...
 
 
@@ -85,20 +90,44 @@ def asset(
     retry_policy: Optional[RetryPolicy] = ...,
     code_version: Optional[str] = ...,
     key: Optional[CoercibleToAssetKey] = None,
-    non_argument_deps: Optional[Union[Set[AssetKey], Set[str]]] = ...,
     check_specs: Optional[Sequence[AssetCheckSpec]] = ...,
     owners: Optional[Sequence[str]] = ...,
+    **kwargs,
 ) -> Callable[[Callable[..., Any]], AssetsDefinition]: ...
 
 
+def _validate_hidden_non_argument_dep_param(
+    non_argument_deps: Any,
+) -> Optional[Union[Set[AssetKey], Set[str]]]:
+    if non_argument_deps is None:
+        return non_argument_deps
+
+    if not isinstance(non_argument_deps, set):
+        check.failed("non_arguments_deps must be a set if not None")
+
+    assert isinstance(non_argument_deps, set)
+
+    check.set_param(non_argument_deps, "non_argument_deps", of_type=(str, AssetKey))
+
+    check.invariant(
+        all(isinstance(dep, str) for dep in non_argument_deps)
+        or all(isinstance(dep, AssetKey) for dep in non_argument_deps),
+    )
+
+    return non_argument_deps
+
+
 @experimental_param(param="resource_defs")
 @experimental_param(param="io_manager_def")
 @experimental_param(param="auto_materialize_policy")
 @experimental_param(param="backfill_policy")
 @experimental_param(param="owners")
 @experimental_param(param="tags")
 @deprecated_param(
-    param="non_argument_deps", breaking_version="2.0.0", additional_warn_text="use `deps` instead."
+    param="non_argument_deps",
+    breaking_version="2.0.0",
+    additional_warn_text="use `deps` instead.",
+    hidden=True,
 )
 def asset(
     compute_fn: Optional[Callable[..., Any]] = None,
@@ -127,9 +156,9 @@ def asset(
     retry_policy: Optional[RetryPolicy] = None,
     code_version: Optional[str] = None,
     key: Optional[CoercibleToAssetKey] = None,
-    non_argument_deps: Optional[Union[Set[AssetKey], Set[str]]] = None,
     check_specs: Optional[Sequence[AssetCheckSpec]] = None,
     owners: Optional[Sequence[str]] = None,
+    **kwargs,
 ) -> Union[AssetsDefinition, Callable[[Callable[..., Any]], AssetsDefinition]]:
     """Create a definition for how to compute an asset.
 
@@ -203,12 +232,13 @@ def asset(
             output when given the same inputs.
         check_specs (Optional[Sequence[AssetCheckSpec]]): (Experimental) Specs for asset checks that
             execute in the decorated function after materializing the asset.
-        non_argument_deps (Optional[Union[Set[AssetKey], Set[str]]]): Deprecated, use deps instead.
-            Set of asset keys that are upstream dependencies, but do not pass an input to the asset.
         key (Optional[CoeercibleToAssetKey]): The key for this asset. If provided, cannot specify key_prefix or name.
         owners (Optional[Sequence[str]]): A list of strings representing owners of the asset. Each
             string can be a user's email address, or a team name prefixed with `team:`,
             e.g. `team:finops`.
+        non_argument_deps (Optional[Union[Set[AssetKey], Set[str]]]): Deprecated, use deps instead.
+            Set of asset keys that are upstream dependencies, but do not pass an input to the asset.
+            Hidden parameter not exposed in the decorator signature, but passed in kwargs.
 
     Examples:
         .. code-block:: python
@@ -220,10 +250,13 @@ def my_asset(my_upstream_asset: int) -> int:
     compute_kind = check.opt_str_param(compute_kind, "compute_kind")
     required_resource_keys = check.opt_set_param(required_resource_keys, "required_resource_keys")
     upstream_asset_deps = _deps_and_non_argument_deps_to_asset_deps(
-        deps=deps, non_argument_deps=non_argument_deps
+        deps=deps,
+        non_argument_deps=_validate_hidden_non_argument_dep_param(kwargs.get("non_argument_deps")),
     )
     resource_defs = dict(check.opt_mapping_param(resource_defs, "resource_defs"))
 
+    only_allow_hidden_params_in_kwargs(asset, kwargs)
+
     args = AssetDecoratorArgs(
         name=name,
         key_prefix=key_prefix,
@@ -463,7 +496,10 @@ def create_assets_def_from_fn_and_decorator_args(
 
 @experimental_param(param="resource_defs")
 @deprecated_param(
-    param="non_argument_deps", breaking_version="2.0.0", additional_warn_text="use `deps` instead."
+    param="non_argument_deps",
+    breaking_version="2.0.0",
+    additional_warn_text="use `deps` instead.",
+    hidden=True,
 )
 def multi_asset(
     *,
@@ -486,8 +522,7 @@ def multi_asset(
     code_version: Optional[str] = None,
     specs: Optional[Sequence[AssetSpec]] = None,
     check_specs: Optional[Sequence[AssetCheckSpec]] = None,
-    # deprecated
-    non_argument_deps: Optional[Union[Set[AssetKey], Set[str]]] = None,
+    **kwargs: Mapping[str, Any],
 ) -> Callable[[Callable[..., Any]], AssetsDefinition]:
     """Create a combined definition of multiple assets that are computed using the same op and same
     upstream assets.
@@ -576,14 +611,19 @@ def my_function(asset0):
     """
     from dagster._core.execution.build_resources import wrap_resources_for_execution
 
+    only_allow_hidden_params_in_kwargs(multi_asset, kwargs)
+
     args = DecoratorAssetsDefinitionBuilderArgs(
         name=name,
         op_description=description,
         specs=check.opt_list_param(specs, "specs", of_type=AssetSpec),
         check_specs_by_output_name=create_check_specs_by_output_name(check_specs),
         asset_out_map=check.opt_mapping_param(outs, "outs", key_type=str, value_type=AssetOut),
         upstream_asset_deps=_deps_and_non_argument_deps_to_asset_deps(
-            deps=deps, non_argument_deps=non_argument_deps
+            deps=deps,
+            non_argument_deps=_validate_hidden_non_argument_dep_param(
+                kwargs.get("non_argument_deps")
+            ),
         ),
         asset_deps=check.opt_mapping_param(
             internal_asset_deps, "internal_asset_deps", key_type=str, value_type=set

python_modules/dagster/dagster_tests/test_annotations.py

@@ -20,6 +20,7 @@
     is_experimental,
     is_experimental_param,
     is_public,
+    only_allow_hidden_params_in_kwargs,
     public,
 )
 from dagster._check import CheckError
@@ -761,3 +762,37 @@ def foo():
 
     dep = next(warning for warning in all_warnings if warning.category == DeprecationWarning)
     assert re.search(r"`[^`]+foo` is deprecated", str(dep.message))
+
+
+def test_hidden_annotations() -> None:
+    @deprecated_param(param="baz", breaking_version="2.0", hidden=True)
+    def with_hidden_args(**kwargs) -> bool:
+        only_allow_hidden_params_in_kwargs(with_hidden_args, kwargs)
+        return True
+
+    assert with_hidden_args(baz="foo")
+    with pytest.raises(
+        TypeError,
+        match="with_hidden_args got an unexpected keyword argument 'does_not_exist'",
+    ):
+        with_hidden_args(does_not_exist="foo")
+
+    with pytest.raises(
+        CheckError,
+        match="Invariant failed. Description: Attempted to mark undefined parameter `baz` deprecated.",
+    ):
+
+        @deprecated_param(param="baz", breaking_version="2.0")
+        def incorrectly_annotated(**kwargs) -> bool:
+            raise NotImplementedError("This function should not be called")
+
+    def vanilla_func(**kwargs) -> bool:
+        only_allow_hidden_params_in_kwargs(vanilla_func, kwargs)
+        return True
+
+    assert vanilla_func()
+    with pytest.raises(
+        TypeError,
+        match="vanilla_func got an unexpected keyword argument 'hidden_param'",
+    ):
+        vanilla_func(hidden_param="foo")