[core-api][experimental] get_asset_provenance (#26684)

## Summary & Motivation

decision: experimental -> public (uncertain)
reason: this has existed for years without change. even if this is not a
commonly-used API, we should probably just make the call that we're not
changing it at this point
docs exist: yes (api only, which is ok)

## How I Tested These Changes

## Changelog

> Insert changelog entry or delete this section.

Author: OwenKephart

python_modules/dagster/dagster/_annotations.py

@@ -1149,3 +1149,6 @@ def only_allow_hidden_params_in_kwargs(annotatable: Annotatable, kwargs: Mapping
             deprecated_params[param].hidden,
             f"Unexpected non-hidden deprecated parameter '{param}' in kwargs. Should never get here.",
         )
+
+
+beta_param = experimental_param  # PLACEHOLDER

python_modules/dagster/dagster/_core/definitions/asset_selection.py

@@ -8,7 +8,7 @@
 from typing_extensions import TypeAlias, TypeGuard
 
 import dagster._check as check
-from dagster._annotations import deprecated, experimental, experimental_param, public
+from dagster._annotations import deprecated, experimental_param, public
 from dagster._core.definitions.asset_check_spec import AssetCheckKey
 from dagster._core.definitions.asset_graph import AssetGraph
 from dagster._core.definitions.asset_key import (
@@ -241,7 +241,7 @@ def groups(*group_strs, include_sources: bool = False) -> "GroupsAssetSelection"
 
     @public
     @staticmethod
-    @experimental
+    @experimental_param(param="include_sources")
     def tag(key: str, value: str, include_sources: bool = False) -> "AssetSelection":
         """Returns a selection that includes materializable assets that have the provided tag, and
         all the asset checks that target them.

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

@@ -16,7 +16,7 @@
 )
 
 import dagster._check as check
-from dagster._annotations import experimental_param, public
+from dagster._annotations import beta_param, experimental_param, public
 from dagster._core.definitions.asset_check_spec import AssetCheckSpec
 from dagster._core.definitions.asset_dep import AssetDep
 from dagster._core.definitions.asset_graph_computation import AssetGraphComputation
@@ -112,7 +112,7 @@ class AssetsDefinition(ResourceAddable, IHasInternalInit):
     _specs_by_key: Mapping[AssetKey, AssetSpec]
     _computation: Optional[AssetGraphComputation]
 
-    @experimental_param(param="execution_type")
+    @beta_param(param="execution_type")
     def __init__(
         self,
         *,
@@ -399,7 +399,7 @@ def __call__(self, *args: object, **kwargs: object) -> object:
         return direct_invocation_result(self, *args, **kwargs)
 
     @public
-    @experimental_param(param="resource_defs")
+    @beta_param(param="resource_defs")
     @staticmethod
     def from_graph(
         graph_def: "GraphDefinition",

python_modules/dagster/dagster/_core/definitions/auto_materialize_policy.py

@@ -3,7 +3,7 @@
 from typing import TYPE_CHECKING, AbstractSet, NamedTuple, Optional  # noqa: UP035
 
 import dagster._check as check
-from dagster._annotations import deprecated, experimental, public
+from dagster._annotations import deprecated, public
 from dagster._serdes.serdes import (
     NamedTupleSerializer,
     UnpackContext,
@@ -55,11 +55,11 @@ class AutoMaterializePolicyType(Enum):
     LAZY = "LAZY"
 
 
-@experimental
 @whitelist_for_serdes(
     old_fields={"time_window_partition_scope_minutes": 1e-6},
     serializer=AutoMaterializePolicySerializer,
 )
+@deprecated(breaking_version="1.10.0")
 class AutoMaterializePolicy(
     NamedTuple(
         "_AutoMaterializePolicy",

python_modules/dagster/dagster/_core/definitions/auto_materialize_rule_impls.py

@@ -4,7 +4,7 @@
 from collections.abc import Iterable, Mapping, Sequence
 from typing import TYPE_CHECKING, AbstractSet, NamedTuple, Optional  # noqa: UP035
 
-from dagster._annotations import experimental
+from dagster._annotations import deprecated
 from dagster._core.asset_graph_view.serializable_entity_subset import SerializableEntitySubset
 from dagster._core.definitions.auto_materialize_rule import AutoMaterializeRule
 from dagster._core.definitions.auto_materialize_rule_evaluation import (
@@ -211,7 +211,7 @@ def evaluate_for_asset(self, context: "AutomationContext") -> "AutomationResult"
 
 
 @whitelist_for_serdes
-@experimental
+@deprecated(breaking_version="1.10.0")
 class AutoMaterializeAssetPartitionsFilter(
     NamedTuple(
         "_AutoMaterializeAssetPartitionsFilter",

python_modules/dagster/dagster/_core/definitions/backfill_policy.py

@@ -3,7 +3,7 @@
 from typing import NamedTuple, Optional
 
 import dagster._check as check
-from dagster._annotations import experimental, public
+from dagster._annotations import beta, public
 from dagster._serdes import whitelist_for_serdes
 from dagster._utils.warnings import disable_dagster_warnings
 
@@ -13,7 +13,7 @@ class BackfillPolicyType(Enum):
     MULTI_RUN = "MULTI_RUN"
 
 
-@experimental
+@beta
 @whitelist_for_serdes
 class BackfillPolicy(
     NamedTuple(

python_modules/dagster/dagster/_core/definitions/data_version.py

@@ -5,7 +5,7 @@
 from typing import TYPE_CHECKING, Callable, Final, NamedTuple, Optional, Union
 
 from dagster import _check as check
-from dagster._annotations import deprecated, experimental
+from dagster._annotations import beta, deprecated
 from dagster._core.loader import LoadingContext
 from dagster._utils.cached_method import cached_method
 
@@ -52,7 +52,7 @@ def __new__(
         )
 
 
-@experimental
+@beta
 class DataVersionsByPartition(
     NamedTuple(
         "_DataVersionsByPartition", [("data_versions_by_partition", Mapping[str, DataVersion])]

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

@@ -2,11 +2,7 @@
 from typing import AbstractSet, Any, Callable, NamedTuple, Optional, Union, overload  # noqa: UP035
 
 import dagster._check as check
-from dagster._annotations import (
-    experimental_param,
-    hidden_param,
-    only_allow_hidden_params_in_kwargs,
-)
+from dagster._annotations import beta_param, hidden_param, only_allow_hidden_params_in_kwargs
 from dagster._config.config_schema import UserConfigSchema
 from dagster._core.definitions.asset_check_spec import AssetCheckSpec
 from dagster._core.definitions.asset_dep import (
@@ -119,9 +115,9 @@ def _validate_hidden_non_argument_dep_param(
     return non_argument_deps
 
 
-@experimental_param(param="resource_defs")
-@experimental_param(param="io_manager_def")
-@experimental_param(param="backfill_policy")
+@beta_param(param="resource_defs")
+@beta_param(param="io_manager_def")
+@beta_param(param="backfill_policy")
 @hidden_param(
     param="non_argument_deps",
     breaking_version="2.0.0",
@@ -539,7 +535,7 @@ def create_assets_def_from_fn_and_decorator_args(
     return builder.create_assets_definition()
 
 
-@experimental_param(param="resource_defs")
+@beta_param(param="resource_defs")
 @hidden_param(
     param="non_argument_deps",
     breaking_version="2.0.0",

python_modules/dagster/dagster/_core/definitions/definitions_class.py

@@ -5,7 +5,7 @@
 from typing_extensions import Self
 
 import dagster._check as check
-from dagster._annotations import deprecated, experimental, public
+from dagster._annotations import deprecated, preview, public
 from dagster._core.definitions.asset_checks import AssetChecksDefinition
 from dagster._core.definitions.asset_graph import AssetGraph
 from dagster._core.definitions.asset_spec import AssetSpec
@@ -46,7 +46,6 @@
 
 
 @public
-@experimental
 def create_repository_using_definitions_args(
     name: str,
     assets: Optional[
@@ -602,7 +601,6 @@ def validate_loadable(defs: "Definitions") -> None:
         defs.get_repository_def().load_all_definitions()
 
     @public
-    @experimental
     @staticmethod
     def merge(*def_sets: "Definitions") -> "Definitions":
         """Merges multiple Definitions objects into a single Definitions object.
@@ -687,13 +685,13 @@ def merge(*def_sets: "Definitions") -> "Definitions":
         )
 
     @public
-    @experimental
+    @preview
     def get_all_asset_specs(self) -> Sequence[AssetSpec]:
         """Returns an AssetSpec object for every asset contained inside the Definitions object."""
         asset_graph = self.get_asset_graph()
         return [asset_node.to_asset_spec() for asset_node in asset_graph.asset_nodes]
 
-    @experimental
+    @preview
     def with_reconstruction_metadata(self, reconstruction_metadata: Mapping[str, str]) -> Self:
         """Add reconstruction metadata to the Definitions object. This is typically used to cache data
         loaded from some external API that is computed during initialization of a code server.

python_modules/dagster/dagster/_core/definitions/events.py

@@ -17,7 +17,7 @@
 from typing_extensions import Self
 
 import dagster._check as check
-from dagster._annotations import PublicAttr, deprecated, experimental_param, public
+from dagster._annotations import PublicAttr, beta_param, deprecated, public
 from dagster._core.definitions.asset_key import (
     AssetKey as AssetKey,
     CoercibleToAssetKey as CoercibleToAssetKey,
@@ -89,7 +89,7 @@ def with_metadata(self, metadata: Optional[Mapping[str, RawMetadataValue]]) -> S
 T = TypeVar("T")
 
 
-@experimental_param(param="data_version")
+@beta_param(param="data_version")
 class Output(Generic[T], EventWithMetadata):
     """Event corresponding to one of an op's outputs.
 

python_modules/dagster/dagster/_core/definitions/partition_mapping.py

@@ -9,7 +9,7 @@
 from typing import NamedTuple, Optional, Union, cast
 
 import dagster._check as check
-from dagster._annotations import PublicAttr, experimental, public
+from dagster._annotations import PublicAttr, beta, public
 from dagster._core.definitions.multi_dimensional_partitions import (
     MultiPartitionKey,
     MultiPartitionsDefinition,
@@ -605,7 +605,7 @@ def get_downstream_partitions_for_partitions(
         return result
 
 
-@experimental
+@beta
 @whitelist_for_serdes
 class MultiToSingleDimensionPartitionMapping(
     BaseMultiPartitionMapping,
@@ -694,7 +694,7 @@ def __new__(
         )
 
 
-@experimental
+@beta
 @whitelist_for_serdes
 class MultiPartitionMapping(
     BaseMultiPartitionMapping,

python_modules/dagster/dagster/_core/definitions/resource_definition.py

@@ -14,7 +14,7 @@
 from typing_extensions import TypeAlias
 
 import dagster._check as check
-from dagster._annotations import experimental_param, public
+from dagster._annotations import beta_param, public
 from dagster._core.decorator_utils import (
     format_docstring_for_description,
     get_function_params,
@@ -54,7 +54,7 @@
 ]
 
 
-@experimental_param(param="version")
+@beta_param(param="version")
 class ResourceDefinition(AnonymousConfigurableDefinition, IHasInternalInit):
     """Core class for defining resources.
 

python_modules/dagster/dagster/_core/definitions/run_status_sensor_definition.py

@@ -9,7 +9,7 @@
 from typing_extensions import TypeAlias
 
 import dagster._check as check
-from dagster._annotations import deprecated_param, experimental_param, public
+from dagster._annotations import beta_param, deprecated_param, public
 from dagster._core.definitions.graph_definition import GraphDefinition
 from dagster._core.definitions.instigation_logger import InstigationLogger
 from dagster._core.definitions.job_definition import JobDefinition
@@ -331,7 +331,7 @@ def get_step_failure_events(self) -> Sequence[DagsterEvent]:
         return [cast(DagsterEvent, record.event_log_entry.dagster_event) for record in records]
 
 
-@experimental_param(param="repository_def")
+@beta_param(param="repository_def")
 def build_run_status_sensor_context(
     sensor_name: str,
     dagster_event: DagsterEvent,

python_modules/dagster/dagster/_core/definitions/source_asset.py

@@ -4,7 +4,7 @@
 from typing_extensions import TypeAlias
 
 import dagster._check as check
-from dagster._annotations import PublicAttr, deprecated, experimental_param, public
+from dagster._annotations import PublicAttr, beta_param, deprecated, deprecated_param, public
 from dagster._core.decorator_utils import get_function_params
 from dagster._core.definitions.asset_spec import AssetExecutionType
 from dagster._core.definitions.data_version import (
@@ -158,9 +158,9 @@ def fn(context: OpExecutionContext) -> Output[None]:
     return DecoratedOpFunction(fn)
 
 
-@experimental_param(param="resource_defs")
-@experimental_param(param="io_manager_def")
-@experimental_param(param="freshness_policy")
+@beta_param(param="resource_defs")
+@beta_param(param="io_manager_def")
+@deprecated_param(param="freshness_policy", breaking_version="1.11.0")
 @deprecated(
     breaking_version="2.0.0",
     additional_warn_text="Use AssetSpec instead. If using the SourceAsset io_manager_key property, "

python_modules/dagster/dagster/_core/execution/context/asset_execution_context.py

@@ -2,7 +2,7 @@
 from typing import AbstractSet, Any, Optional  # noqa: UP035
 
 import dagster._check as check
-from dagster._annotations import deprecated, experimental, public
+from dagster._annotations import deprecated, public
 from dagster._core.definitions.asset_check_spec import AssetCheckKey
 from dagster._core.definitions.assets import AssetsDefinition
 from dagster._core.definitions.data_version import DataProvenance, DataVersion
@@ -558,7 +558,6 @@ def selected_asset_check_keys(self) -> AbstractSet[AssetCheckKey]:
     #### data lineage related
 
     @public
-    @experimental
     @_copy_docs_from_op_execution_context
     def get_asset_provenance(self, asset_key: AssetKey) -> Optional[DataProvenance]:
         return self.op_execution_context.get_asset_provenance(asset_key=asset_key)

python_modules/dagster/dagster/_core/execution/context/op_execution_context.py

@@ -3,7 +3,7 @@
 from typing import AbstractSet, Any, Optional, cast  # noqa: UP035
 
 import dagster._check as check
-from dagster._annotations import deprecated, experimental, public
+from dagster._annotations import deprecated, public
 from dagster._core.definitions.asset_check_spec import AssetCheckKey
 from dagster._core.definitions.assets import AssetsDefinition
 from dagster._core.definitions.data_version import (
@@ -1211,7 +1211,6 @@ def self_dependent_asset(context: AssetExecutionContext, self_dependent_asset):
         return self._step_execution_context.asset_partitions_time_window_for_input(input_name)
 
     @public
-    @experimental
     def get_asset_provenance(self, asset_key: AssetKey) -> Optional[DataProvenance]:
         """Return the provenance information for the most recent materialization of an asset.
 

python_modules/dagster/dagster/_core/storage/fs_io_manager.py

@@ -9,7 +9,7 @@
     DagsterInvariantViolationError,
     Field as DagsterField,
 )
-from dagster._annotations import experimental
+from dagster._annotations import superseded
 from dagster._config import StringSource
 from dagster._config.pythonic_config import ConfigurableIOManagerFactory
 from dagster._core.definitions.events import AssetKey, AssetMaterialization
@@ -137,6 +137,7 @@ def create_io_manager(self, context: InitResourceContext) -> "PickledObjectFiles
     config_schema=FilesystemIOManager.to_config_schema(),
     description="Built-in filesystem IO manager that stores and retrieves values using pickling.",
 )
+@superseded(additional_warn_text="Use FilesystemIOManager directly instead")
 def fs_io_manager(init_context: InitResourceContext) -> "PickledObjectFilesystemIOManager":
     """Built-in filesystem IO manager that stores and retrieves values using pickling.
 
@@ -339,7 +340,6 @@ def load_input(self, context: InputContext) -> object:
 
 @dagster_maintained_io_manager
 @io_manager(config_schema={"base_dir": DagsterField(StringSource, is_required=True)})
-@experimental
 def custom_path_fs_io_manager(
     init_context: InitResourceContext,
 ) -> CustomPathPickledObjectFilesystemIOManager: