mark as beta/experimental

Author: anuthebananu

python_modules/dagster/dagster/_core/definitions/asset_graph.py

@@ -157,6 +157,7 @@ def execution_set_entity_keys(self) -> AbstractSet[EntityKey]:
 
     @property
     def new_freshness_policy(self) -> Optional[NewFreshnessPolicy]:
+        """Experimental, do not use."""
         return self._spec.new_freshness_policy
 
     ##### ASSET GRAPH SPECIFIC INTERFACE

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

@@ -212,6 +212,7 @@ def kinds(self) -> Optional[set[str]]:
 
     @property
     def new_freshness_policy(self) -> Optional[NewFreshnessPolicy]:
+        """Experimental, do not use."""
         return self._spec.new_freshness_policy
 
     def to_out(self) -> Out:

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

@@ -16,6 +16,7 @@
 import dagster._check as check
 from dagster._annotations import (
     PublicAttr,
+    beta_param,
     hidden_param,
     only_allow_hidden_params_in_kwargs,
     public,
@@ -99,6 +100,10 @@ def validate_kind_tags(kinds: Optional[AbstractSet[str]]) -> None:
         raise DagsterInvalidDefinitionError("Assets can have at most three kinds currently.")
 
 
+@beta_param(
+    param="new_freshness_policy",
+    additional_warn_text="Currently experimental. Use freshness checks instead to define asset freshness.",
+)
 @hidden_param(
     param="freshness_policy",
     breaking_version="1.10.0",
@@ -141,7 +146,7 @@ class AssetSpec(IHasInternalInit, IHaveNew, LegacyNamedTupleMixin):
             will be made visible in the Dagster UI.
         partitions_def (Optional[PartitionsDefinition]): Defines the set of partition keys that
             compose the asset.
-        new_freshness_policy (Optional[NewFreshnessPolicy]): A condition that delineates when an asset is considered fresh. Use this over `freshness_policy`.
+        new_freshness_policy (Optional[NewFreshnessPolicy]): A condition that delineates when an asset is considered fresh. Currently experimental.
     """
 
     key: PublicAttr[AssetKey]
@@ -302,6 +307,7 @@ def with_io_manager_key(self, io_manager_key: str) -> "AssetSpec":
         )
 
     @public
+    @beta_param(param="new_freshness_policy", additional_warn_text="Currently experimental.")
     def replace_attributes(
         self,
         *,

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

@@ -115,6 +115,7 @@ class AssetsDefinition(ResourceAddable, IHasInternalInit):
     _computation: Optional[AssetGraphComputation]
 
     @beta_param(param="execution_type")
+    @beta_param(param="new_freshness_policy_by_key")
     def __init__(
         self,
         *,
@@ -406,6 +407,7 @@ def __call__(self, *args: object, **kwargs: object) -> object:
 
     @public
     @beta_param(param="resource_defs")
+    @beta_param(param="new_freshness_policy_by_key")
     @staticmethod
     def from_graph(
         graph_def: "GraphDefinition",
@@ -498,7 +500,8 @@ def from_graph(
                 owners to be associated with each of the asset keys for this node.
             new_freshness_policy_by_output_name (Optional[Mapping[str, Optional[NewFreshnessPolicy]]]): Defines asset
                 freshness conditions to be associated with some or all of the output assets for this node.
-                This is intended to replace the existing, deprecated `FreshnessPolicy` construct.
+                Currently experimental, it is intended to replace the existing, deprecated `FreshnessPolicy` construct.
+                For now, continue to use freshness checks to define asset freshness.
         """
         return AssetsDefinition._from_node(
             node_def=graph_def,

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

@@ -253,8 +253,7 @@ def asset(
         pool (Optional[str]): A string that identifies the concurrency pool that governs this asset's execution.
         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.
-        new_freshness_policy (Optional[NewFreshnessPolicy]): A condition that must be met for an asset to be considered fresh.
+            Hidden parameter not exposed in the decorator signature, but passed in kwargs. Currently experimental.
 
     Examples:
         .. code-block:: python
@@ -710,7 +709,7 @@ def my_function(asset0):
         decorator_name="@multi_asset",
         execution_type=AssetExecutionType.MATERIALIZATION,
         pool=pool,
-        new_freshness_policy=None,  # TODO enable new_freshness_policy for @multi_asset
+        new_freshness_policy=None,
     )
 
     def inner(fn: Callable[..., Any]) -> AssetsDefinition:

python_modules/dagster/dagster/_core/definitions/freshness.py

@@ -1,12 +1,12 @@
 from enum import Enum
 from typing import NamedTuple
 
+from dagster._annotations import beta
 from dagster._core.definitions.asset_key import AssetKey
 from dagster._serdes import whitelist_for_serdes
 
 
-# TODO finalize states and naming
-# https://dagsterlabs.slack.com/archives/C047L6H0LF4/p1743697967200509
+@beta
 @whitelist_for_serdes
 class FreshnessState(str, Enum):
     PASSING = "PASSING"
@@ -15,6 +15,7 @@ class FreshnessState(str, Enum):
     UNKNOWN = "UNKNOWN"
 
 
+@beta
 @whitelist_for_serdes
 class FreshnessStateEvaluation(
     NamedTuple(

python_modules/dagster/dagster/_core/definitions/new_freshness_policy.py

@@ -3,6 +3,7 @@
 from typing import Optional
 
 from dagster import _check as check
+from dagster._annotations import beta
 from dagster._record import IHaveNew, record_custom
 from dagster._serdes import whitelist_for_serdes
 
@@ -27,10 +28,11 @@ class NewFreshnessPolicy(ABC):
     def policy_type(self) -> FreshnessPolicyType: ...
 
 
+@beta
 @whitelist_for_serdes
 @record_custom
 class TimeWindowFreshnessPolicy(NewFreshnessPolicy, IHaveNew):
-    """A freshness condition that considers an asset fresh if it was materialized within a specified time window (time_window_minutes).
+    """[BETA/EXPERIMENTAL] A freshness condition that considers an asset fresh if it was materialized within a specified time window (time_window_minutes).
 
     The asset is considered stale if its last materialization is older than time_window_minutes, and in a warning state
     if its last materialization is older than the warning time window but newer than the critical time window.

python_modules/dagster/dagster/_core/definitions/remote_asset_graph.py

@@ -130,6 +130,7 @@ def job_names(self) -> Sequence[str]:
 
     @property
     def new_freshness_policy(self) -> Optional[NewFreshnessPolicy]:
+        """Experimental, do not use."""
         return self.resolve_to_singular_repo_scoped_node().asset_node_snap.new_freshness_policy
 
 
@@ -206,6 +207,7 @@ def pools(self) -> Optional[set[str]]:
 
     @property
     def new_freshness_policy(self) -> Optional[NewFreshnessPolicy]:
+        """Experimental, do not use."""
         return self.asset_node_snap.new_freshness_policy