[core-api][experimental] AutomationCondition.any_downstream_conditions (

#26707)

## Summary & Motivation

decision: experimental -> beta
reason: we might want to subtly change the default behavior of this
(e.g. how far down the tree it looks)
docs exist: n/a

## How I Tested These Changes

## Changelog

> Insert changelog entry or delete this section.

python_modules/dagster/dagster/_core/definitions/asset_check_factories/freshness_checks/last_update.py

@@ -2,7 +2,7 @@
 from typing import Any, Dict, Iterable, Optional, Sequence, Union, cast
 
 from dagster import _check as check
-from dagster._annotations import experimental
+from dagster._annotations import beta
 from dagster._core.definitions.asset_check_factories.utils import (
     DEADLINE_CRON_PARAM_KEY,
     DEFAULT_FRESHNESS_SEVERITY,
@@ -40,7 +40,7 @@
 )
 
 
-@experimental
+@beta
 def build_last_update_freshness_checks(
     *,
     assets: Sequence[Union[CoercibleToAssetKey, AssetsDefinition, SourceAsset]],

python_modules/dagster/dagster/_core/definitions/asset_check_factories/freshness_checks/sensor.py

@@ -2,7 +2,7 @@
 from typing import Iterator, Optional, Sequence, Tuple, Union, cast
 
 from dagster import _check as check
-from dagster._annotations import experimental
+from dagster._annotations import beta
 from dagster._core.definitions.asset_check_factories.utils import (
     FRESH_UNTIL_METADATA_KEY,
     ensure_no_duplicate_asset_checks,
@@ -33,7 +33,7 @@
     """
 
 
-@experimental
+@beta
 def build_sensor_for_freshness_checks(
     *,
     freshness_checks: Sequence[AssetChecksDefinition],

python_modules/dagster/dagster/_core/definitions/asset_check_factories/freshness_checks/time_partition.py

@@ -1,7 +1,7 @@
 from typing import Any, Dict, Iterable, Sequence, Union
 
 from dagster import _check as check
-from dagster._annotations import experimental
+from dagster._annotations import beta
 from dagster._core.definitions.asset_check_factories.utils import (
     DEADLINE_CRON_PARAM_KEY,
     DEFAULT_FRESHNESS_SEVERITY,
@@ -37,7 +37,7 @@
 )
 
 
-@experimental
+@beta
 def build_time_partition_freshness_checks(
     *,
     assets: Sequence[Union[SourceAsset, CoercibleToAssetKey, AssetsDefinition]],

python_modules/dagster/dagster/_core/definitions/asset_check_factories/metadata_bounds_checks.py

@@ -2,7 +2,7 @@
 from typing import Optional, Sequence, Tuple, Union, cast
 
 import dagster._check as check
-from dagster._annotations import experimental
+from dagster._annotations import beta
 from dagster._core.definitions.asset_check_factories.utils import (
     assets_to_keys,
     build_multi_asset_check,
@@ -21,7 +21,7 @@
 from dagster._core.instance import DagsterInstance
 
 
-@experimental
+@beta
 def build_metadata_bounds_checks(
     *,
     assets: Sequence[Union[CoercibleToAssetKey, AssetsDefinition, SourceAsset]],

python_modules/dagster/dagster/_core/definitions/asset_check_factories/schema_change_checks.py

@@ -2,7 +2,7 @@
 
 from pydantic import BaseModel
 
-from dagster._annotations import experimental
+from dagster._annotations import beta
 from dagster._core.definitions.asset_check_factories.utils import build_multi_asset_check
 from dagster._core.definitions.asset_check_spec import (
     AssetCheckKey,
@@ -17,7 +17,7 @@
 from dagster._core.instance import DagsterInstance
 
 
-@experimental
+@beta
 def build_column_schema_change_checks(
     *,
     assets: Sequence[Union[CoercibleToAssetKey, AssetsDefinition, SourceAsset]],

python_modules/dagster/dagster/_core/definitions/asset_out.py

@@ -1,12 +1,7 @@
 from typing import Any, Mapping, Optional, Sequence, Type, Union
 
 import dagster._check as check
-from dagster._annotations import (
-    experimental_param,
-    hidden_param,
-    only_allow_hidden_params_in_kwargs,
-    public,
-)
+from dagster._annotations import hidden_param, only_allow_hidden_params_in_kwargs, public
 from dagster._core.definitions.asset_dep import AssetDep
 from dagster._core.definitions.asset_spec import AssetSpec
 from dagster._core.definitions.auto_materialize_policy import AutoMaterializePolicy
@@ -35,8 +30,6 @@
 EMPTY_ASSET_KEY_SENTINEL = AssetKey([])
 
 
-@experimental_param(param="owners")
-@experimental_param(param="tags")
 @hidden_param(
     param="freshness_policy",
     breaking_version="1.10.0",

python_modules/dagster/dagster/_core/definitions/asset_spec.py

@@ -18,7 +18,6 @@
 import dagster._check as check
 from dagster._annotations import (
     PublicAttr,
-    experimental_param,
     hidden_param,
     only_allow_hidden_params_in_kwargs,
     public,
@@ -96,7 +95,6 @@ def validate_kind_tags(kinds: Optional[AbstractSet[str]]) -> None:
         raise DagsterInvalidDefinitionError("Assets can have at most three kinds currently.")
 
 
-@experimental_param(param="owners")
 @hidden_param(
     param="freshness_policy",
     breaking_version="1.10.0",

python_modules/dagster/dagster/_core/definitions/assets.py

@@ -119,7 +119,6 @@ class AssetsDefinition(ResourceAddable, IHasInternalInit):
     _specs_by_key: Mapping[AssetKey, AssetSpec]
     _computation: Optional[AssetGraphComputation]
 
-    @experimental_param(param="specs")
     @experimental_param(param="execution_type")
     def __init__(
         self,

python_modules/dagster/dagster/_core/definitions/automation_condition_sensor_definition.py

@@ -2,7 +2,7 @@
 from typing import Any, Mapping, Optional, cast
 
 import dagster._check as check
-from dagster._annotations import experimental
+from dagster._annotations import alpha_param, beta_param
 from dagster._core.definitions.asset_selection import AssetSelection, CoercibleToAssetSelection
 from dagster._core.definitions.declarative_automation.automation_condition import (
     AutomationCondition,
@@ -74,7 +74,8 @@ def not_supported(context) -> None:
     )
 
 
-@experimental
+@beta_param(param="use_user_code_server")
+@alpha_param(param="default_condition")
 class AutomationConditionSensorDefinition(SensorDefinition):
     """Targets a set of assets and repeatedly evaluates all the AutomationConditions on all of
     those assets to determine which to request runs for.

python_modules/dagster/dagster/_core/definitions/declarative_automation/automation_condition.py

@@ -6,7 +6,7 @@
 from typing_extensions import Self
 
 import dagster._check as check
-from dagster._annotations import experimental, public
+from dagster._annotations import beta, public
 from dagster._core.asset_graph_view.entity_subset import EntitySubset
 from dagster._core.asset_graph_view.serializable_entity_subset import SerializableEntitySubset
 from dagster._core.definitions.asset_key import (
@@ -632,7 +632,7 @@ def on_missing() -> "AndAutomationCondition":
         ).with_label("on_missing")
 
     @public
-    @experimental
+    @beta
     @staticmethod
     def any_downstream_conditions() -> "BuiltinAutomationCondition":
         """Returns an AutomationCondition which represents the union of all distinct downstream conditions."""

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

@@ -134,7 +134,6 @@ def _validate_hidden_non_argument_dep_param(
 @experimental_param(param="resource_defs")
 @experimental_param(param="io_manager_def")
 @experimental_param(param="backfill_policy")
-@experimental_param(param="owners")
 @hidden_param(
     param="non_argument_deps",
     breaking_version="2.0.0",
@@ -736,7 +735,6 @@ def graph_asset(
 ) -> Callable[[Callable[..., Any]], AssetsDefinition]: ...
 
 
-@experimental_param(param="owners")
 @hidden_param(
     param="freshness_policy",
     breaking_version="1.10.0",

python_modules/dagster/dagster/_core/definitions/external_asset.py

@@ -1,7 +1,7 @@
 from typing import List, Sequence
 
 from dagster import _check as check
-from dagster._annotations import deprecated, experimental
+from dagster._annotations import deprecated
 from dagster._core.definitions.asset_spec import (
     SYSTEM_METADATA_KEY_AUTO_OBSERVE_INTERVAL_MINUTES,
     SYSTEM_METADATA_KEY_IO_MANAGER_KEY,
@@ -23,7 +23,6 @@ def external_asset_from_spec(spec: AssetSpec) -> AssetsDefinition:
 
 
 @deprecated(breaking_version="1.9.0", additional_warn_text="Directly use the AssetSpecs instead.")
-@experimental
 def external_assets_from_specs(specs: Sequence[AssetSpec]) -> List[AssetsDefinition]:
     """Create an external assets definition from a sequence of asset specs.
 

python_modules/dagster/dagster/_core/definitions/input.py

@@ -14,7 +14,7 @@
 )
 
 import dagster._check as check
-from dagster._annotations import PublicAttr, deprecated_param, experimental_param
+from dagster._annotations import PublicAttr, deprecated_param, superseded
 from dagster._core.definitions.events import AssetKey
 from dagster._core.definitions.inference import InferredInputProps
 from dagster._core.definitions.metadata import (
@@ -59,8 +59,7 @@ def _check_default_value(input_name: str, dagster_type: DagsterType, default_val
     return default_value  # type: ignore  # (pyright bug)
 
 
-@experimental_param(param="asset_key")
-@experimental_param(param="asset_partitions")
+@superseded(additional_warn_text="Use `In` instead", emit_runtime_warning=False)
 class InputDefinition:
     """Defines an argument to an op's compute function.
 

python_modules/dagster/dagster/_core/definitions/metadata/source_code.py

@@ -6,7 +6,7 @@
 from typing import TYPE_CHECKING, Any, Callable, List, Optional, Sequence, Union
 
 import dagster._check as check
-from dagster._annotations import experimental, public
+from dagster._annotations import beta, public
 from dagster._core.definitions.metadata.metadata_set import (
     NamespacedMetadataSet as NamespacedMetadataSet,
     TableMetadataSet as TableMetadataSet,
@@ -22,7 +22,7 @@
 DEFAULT_SOURCE_FILE_KEY = "asset_definition"
 
 
-@experimental
+@beta
 @whitelist_for_serdes
 class LocalFileCodeReference(DagsterModel):
     """Represents a local file source location."""
@@ -32,7 +32,7 @@ class LocalFileCodeReference(DagsterModel):
     label: Optional[str] = None
 
 
-@experimental
+@beta
 @whitelist_for_serdes
 class UrlCodeReference(DagsterModel):
     """Represents a source location which points at a URL, for example
@@ -43,7 +43,7 @@ class UrlCodeReference(DagsterModel):
     label: Optional[str] = None
 
 
-@experimental
+@beta
 @whitelist_for_serdes
 class CodeReferencesMetadataValue(DagsterModel, MetadataValue["CodeReferencesMetadataValue"]):
     """Metadata value type which represents source locations (locally or otherwise)
@@ -148,7 +148,7 @@ def _with_code_source_single_definition(
     )
 
 
-@experimental
+@beta
 class FilePathMapping(ABC):
     """Base class which defines a file path mapping function. These functions are used to map local file paths
     to their corresponding paths in a source control repository.
@@ -172,7 +172,7 @@ def convert_to_source_control_path(self, local_path: Path) -> str:
         """
 
 
-@experimental
+@beta
 @dataclass
 class AnchorBasedFilePathMapping(FilePathMapping):
     """Specifies the mapping between local file paths and their corresponding paths in a source control repository,
@@ -291,7 +291,7 @@ def _build_gitlab_url(url: str, branch: str) -> str:
     return f"{url}/-/tree/{branch}"
 
 
-@experimental
+@beta
 def link_code_references_to_git(
     assets_defs: Sequence[
         Union["AssetsDefinition", "SourceAsset", "CacheableAssetsDefinition", "AssetSpec"]
@@ -353,7 +353,7 @@ def link_code_references_to_git(
     ]
 
 
-@experimental
+@beta
 def with_source_code_references(
     assets_defs: Sequence[
         Union["AssetsDefinition", "SourceAsset", "CacheableAssetsDefinition", "AssetSpec"]

python_modules/dagster/dagster/_core/definitions/multi_asset_sensor_definition.py

@@ -18,7 +18,7 @@
 )
 
 import dagster._check as check
-from dagster._annotations import deprecated_param, experimental, public
+from dagster._annotations import deprecated_param, public, superseded
 from dagster._core.definitions.asset_selection import AssetSelection
 from dagster._core.definitions.assets import AssetsDefinition
 from dagster._core.definitions.events import AssetKey
@@ -168,7 +168,6 @@ def get_stringified_cursor(self) -> str:
     breaking_version="2.0",
     additional_warn_text="Use `last_tick_completion_time` instead.",
 )
-@experimental
 class MultiAssetSensorEvaluationContext(SensorEvaluationContext):
     """The context object available as the argument to the evaluation function of a
     :py:class:`dagster.MultiAssetSensorDefinition`.
@@ -984,7 +983,6 @@ def get_cursor_from_latest_materializations(
     return cursor_str
 
 
-@experimental
 def build_multi_asset_sensor_context(
     *,
     monitored_assets: Union[Sequence[AssetKey], AssetSelection],
@@ -1103,7 +1101,12 @@ def build_multi_asset_sensor_context(
 ]
 
 
-@experimental
+@superseded(
+    additional_warn_text="For most use cases, Declarative Automation should be used instead of "
+    "multi_asset_sensors to monitor the status of upstream assets and launch runs in response. "
+    "In cases where side effects are required, or a specific job must be targeted for execution, "
+    "multi_asset_sensors may be used."
+)
 class MultiAssetSensorDefinition(SensorDefinition):
     """Define an asset sensor that initiates a set of runs based on the materialization of a list of
     assets.

python_modules/dagster/dagster/_core/definitions/reconstruct.py

@@ -25,7 +25,6 @@
 
 import dagster._check as check
 import dagster._seven as seven
-from dagster._annotations import experimental
 from dagster._core.code_pointer import (
     CodePointer,
     CustomPointer,
@@ -449,7 +448,6 @@ def make_bar_job():
     return bootstrap_standalone_recon_job(pointer)
 
 
-@experimental
 def build_reconstructable_job(
     reconstructor_module_name: str,
     reconstructor_function_name: str,

python_modules/dagster/dagster/_core/definitions/result.py

@@ -1,7 +1,7 @@
 from typing import Mapping, NamedTuple, Optional, Sequence
 
 import dagster._check as check
-from dagster._annotations import PublicAttr, experimental
+from dagster._annotations import PublicAttr
 from dagster._core.definitions.asset_check_result import AssetCheckResult
 from dagster._core.definitions.data_version import DataVersion
 from dagster._core.definitions.events import AssetKey, CoercibleToAssetKey
@@ -74,7 +74,6 @@ class MaterializeResult(AssetResult):
     """
 
 
-@experimental
 class ObserveResult(AssetResult):
     """An object representing a successful observation of an asset. These can be returned from an
     @observable_source_asset decorated function to pass metadata.

python_modules/dagster/dagster/_core/definitions/run_request.py

@@ -16,7 +16,7 @@
 )
 
 import dagster._check as check
-from dagster._annotations import PublicAttr, experimental_param
+from dagster._annotations import PublicAttr
 from dagster._core.definitions.asset_check_evaluation import AssetCheckEvaluation
 from dagster._core.definitions.asset_check_spec import AssetCheckKey
 from dagster._core.definitions.asset_graph_subset import AssetGraphSubset
@@ -354,9 +354,6 @@ def __new__(
         )
 
 
-@experimental_param(
-    param="asset_events", additional_warn_text="Runless asset events are experimental"
-)
 class SensorResult(
     NamedTuple(
         "_SensorResult",