remove new_freshness_policy everywhere

Author: anuthebananu

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

@@ -24,7 +24,6 @@
 from dagster._core.definitions.events import AssetKey
 from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.metadata import ArbitraryMetadataMapping
-from dagster._core.definitions.new_freshness_policy import NewFreshnessPolicy
 from dagster._core.definitions.partition import PartitionsDefinition
 from dagster._core.definitions.partition_mapping import PartitionMapping
 from dagster._core.definitions.resolved_asset_deps import ResolvedAssetDependencies
@@ -155,11 +154,6 @@ def execution_set_entity_keys(self) -> AbstractSet[EntityKey]:
         else:
             return self.assets_def.asset_and_check_keys
 
-    @property
-    def new_freshness_policy(self) -> Optional[NewFreshnessPolicy]:
-        """Experimental, do not use."""
-        return self._spec.new_freshness_policy
-
     ##### ASSET GRAPH SPECIFIC INTERFACE
 
     @property

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

@@ -17,7 +17,6 @@
 )
 from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.input import NoValueSentinel
-from dagster._core.definitions.new_freshness_policy import NewFreshnessPolicy
 from dagster._core.definitions.output import Out
 from dagster._core.definitions.partition import PartitionsDefinition
 from dagster._core.definitions.utils import resolve_automation_condition
@@ -210,11 +209,6 @@ def tags(self) -> Optional[Mapping[str, str]]:
     def kinds(self) -> Optional[set[str]]:
         return self._spec.kinds
 
-    @property
-    def new_freshness_policy(self) -> Optional[NewFreshnessPolicy]:
-        """Experimental, do not use."""
-        return self._spec.new_freshness_policy
-
     def to_out(self) -> Out:
         return Out(
             dagster_type=self.dagster_type,
@@ -231,15 +225,13 @@ def to_spec(
         deps: Sequence[AssetDep],
         additional_tags: Mapping[str, str] = {},
         partitions_def: Optional[PartitionsDefinition] = ...,
-        new_freshness_policy: Optional[NewFreshnessPolicy] = ...,
     ) -> AssetSpec:
         return self._spec.replace_attributes(
             key=key,
             tags={**additional_tags, **self.tags} if self.tags else additional_tags,
             kinds=self.kinds,
             deps=[*self._spec.deps, *deps],
             partitions_def=partitions_def if partitions_def is not None else ...,
-            new_freshness_policy=new_freshness_policy if new_freshness_policy is not None else ...,
         )
 
     @public

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

@@ -16,7 +16,6 @@
 import dagster._check as check
 from dagster._annotations import (
     PublicAttr,
-    beta_param,
     hidden_param,
     only_allow_hidden_params_in_kwargs,
     public,
@@ -27,11 +26,11 @@
     AutomationCondition,
 )
 from dagster._core.definitions.events import AssetKey, CoercibleToAssetKey
-from dagster._core.definitions.freshness_policy import FreshnessPolicy
-from dagster._core.definitions.new_freshness_policy import (
+from dagster._core.definitions.freshness import (
     INTERNAL_FRESHNESS_POLICY_METADATA_KEY,
-    NewFreshnessPolicy,
+    InternalFreshnessPolicy,
 )
+from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.partition import PartitionsDefinition
 from dagster._core.definitions.partition_mapping import PartitionMapping
 from dagster._core.definitions.utils import (
@@ -103,10 +102,6 @@ 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",
@@ -154,7 +149,6 @@ 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. Currently experimental.
     """
 
     key: PublicAttr[AssetKey]
@@ -169,7 +163,6 @@ class AssetSpec(IHasInternalInit, IHaveNew, LegacyNamedTupleMixin):
     owners: PublicAttr[Sequence[str]]
     tags: PublicAttr[Mapping[str, str]]
     partitions_def: PublicAttr[Optional[PartitionsDefinition]]
-    new_freshness_policy: Optional[NewFreshnessPolicy]
 
     def __new__(
         cls,
@@ -186,7 +179,6 @@ def __new__(
         tags: Optional[Mapping[str, str]] = None,
         kinds: Optional[set[str]] = None,
         partitions_def: Optional[PartitionsDefinition] = None,
-        new_freshness_policy: Optional[NewFreshnessPolicy] = None,
         **kwargs,
     ):
         from dagster._core.definitions.asset_dep import coerce_to_deps_and_check_duplicates
@@ -212,7 +204,7 @@ def __new__(
         }
         validate_kind_tags(kind_tags)
 
-        internal_freshness_policy: Optional[NewFreshnessPolicy] = kwargs.get(
+        internal_freshness_policy: Optional[InternalFreshnessPolicy] = kwargs.get(
             "internal_freshness_policy"
         )
         if internal_freshness_policy:
@@ -256,9 +248,6 @@ def __new__(
             partitions_def=check.opt_inst_param(
                 partitions_def, "partitions_def", PartitionsDefinition
             ),
-            new_freshness_policy=check.opt_inst_param(
-                new_freshness_policy, "new_freshness_policy", NewFreshnessPolicy
-            ),
         )
 
     @staticmethod
@@ -293,7 +282,6 @@ def dagster_internal_init(
             tags=tags,
             kinds=kinds,
             partitions_def=partitions_def,
-            new_freshness_policy=kwargs.get("new_freshness_policy"),
         )
 
     @cached_property
@@ -333,7 +321,6 @@ 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,
         *,
@@ -350,7 +337,6 @@ def replace_attributes(
         kinds: Optional[set[str]] = ...,
         partitions_def: Optional[PartitionsDefinition] = ...,
         freshness_policy: Optional[FreshnessPolicy] = ...,
-        new_freshness_policy: Optional[NewFreshnessPolicy] = ...,
     ) -> "AssetSpec":
         """Returns a new AssetSpec with the specified attributes replaced."""
         current_tags_without_kinds = {
@@ -375,9 +361,6 @@ def replace_attributes(
                 tags=tags if tags is not ... else current_tags_without_kinds,
                 kinds=kinds if kinds is not ... else self.kinds,
                 partitions_def=partitions_def if partitions_def is not ... else self.partitions_def,
-                new_freshness_policy=new_freshness_policy
-                if new_freshness_policy is not ...
-                else self.new_freshness_policy,
             )
 
     @public
@@ -425,7 +408,6 @@ def merge_attributes(
                 tags={**current_tags_without_kinds, **(tags if tags is not ... else {})},
                 kinds={*self.kinds, *(kinds if kinds is not ... else {})},
                 partitions_def=self.partitions_def,
-                new_freshness_policy=self.new_freshness_policy,
             )
 
 
@@ -482,6 +464,6 @@ def map_asset_specs(
     ]
 
 
-def attach_internal_freshness_policy(spec: AssetSpec, policy: NewFreshnessPolicy) -> AssetSpec:
+def attach_internal_freshness_policy(spec: AssetSpec, policy: InternalFreshnessPolicy) -> AssetSpec:
     """Apply a freshness policy to an asset spec, attaching it to the spec's metadata."""
     return spec.merge_attributes(metadata={INTERNAL_FRESHNESS_POLICY_METADATA_KEY: str(policy)})

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

@@ -38,7 +38,6 @@
 from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.metadata import ArbitraryMetadataMapping
 from dagster._core.definitions.multi_dimensional_partitions import MultiPartitionsDefinition
-from dagster._core.definitions.new_freshness_policy import NewFreshnessPolicy
 from dagster._core.definitions.node_definition import NodeDefinition
 from dagster._core.definitions.op_definition import OpDefinition
 from dagster._core.definitions.op_invocation import direct_invocation_result
@@ -103,7 +102,6 @@ class AssetsDefinition(ResourceAddable, IHasInternalInit):
         "asset_deps",
         "owners_by_key",
         "partitions_def",
-        "new_freshness_policy_by_key",
     }
 
     # partition mappings are also tracked inside the AssetSpecs, but this enables faster access by
@@ -152,7 +150,6 @@ def __init__(
         auto_materialize_policies_by_key: Optional[Mapping[AssetKey, AutoMaterializePolicy]] = None,
         # if adding new fields, make sure to handle them in the with_attributes, from_graph,
         # from_op, and get_attributes_dict methods
-        new_freshness_policy_by_key: Optional[Mapping[AssetKey, NewFreshnessPolicy]] = None,
     ):
         from dagster._core.definitions.graph_definition import GraphDefinition
         from dagster._core.execution.build_resources import wrap_resources_for_execution
@@ -241,7 +238,6 @@ def __init__(
             check.invariant(partition_mappings is None)
             check.invariant(asset_deps is None)
             check.invariant(partitions_def is None)
-            check.invariant(new_freshness_policy_by_key is None)
             resolved_specs = specs
 
         else:
@@ -282,7 +278,6 @@ def __init__(
                 descriptions_by_key=descriptions_by_key,
                 code_versions_by_key=None,
                 partitions_def=partitions_def,
-                new_freshness_policy_by_key=new_freshness_policy_by_key,
             )
 
         normalized_specs: list[AssetSpec] = []
@@ -406,7 +401,6 @@ def __call__(self, *args: object, **kwargs: object) -> object:
 
     @public
     @beta_param(param="resource_defs")
-    @beta_param(param="new_freshness_policy_by_output_name")
     @staticmethod
     def from_graph(
         graph_def: "GraphDefinition",
@@ -436,9 +430,6 @@ def from_graph(
         auto_materialize_policies_by_output_name: Optional[
             Mapping[str, Optional[AutoMaterializePolicy]]
         ] = None,
-        new_freshness_policy_by_output_name: Optional[
-            Mapping[str, Optional[NewFreshnessPolicy]]
-        ] = None,
     ) -> "AssetsDefinition":
         """Constructs an AssetsDefinition from a GraphDefinition.
 
@@ -497,10 +488,6 @@ def from_graph(
             backfill_policy (Optional[BackfillPolicy]): Defines this asset's BackfillPolicy
             owners_by_key (Optional[Mapping[AssetKey, Sequence[str]]]): Defines
                 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.
-                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,
@@ -526,7 +513,6 @@ def from_graph(
             check_specs=check_specs,
             owners_by_output_name=owners_by_output_name,
             code_versions_by_output_name=code_versions_by_output_name,
-            new_freshness_policy_by_output_name=new_freshness_policy_by_output_name,
         )
 
     @public
@@ -655,9 +641,6 @@ def _from_node(
         can_subset: bool = False,
         check_specs: Optional[Sequence[AssetCheckSpec]] = None,
         owners_by_output_name: Optional[Mapping[str, Sequence[str]]] = None,
-        new_freshness_policy_by_output_name: Optional[
-            Mapping[str, Optional[NewFreshnessPolicy]]
-        ] = None,
     ) -> "AssetsDefinition":
         from dagster._core.definitions.decorators.decorator_assets_definition_builder import (
             _validate_check_specs_target_relevant_asset_keys,
@@ -762,9 +745,6 @@ def _output_dict_to_asset_dict(
             descriptions_by_key=_output_dict_to_asset_dict(descriptions_by_output_name),
             code_versions_by_key=_output_dict_to_asset_dict(code_versions_by_output_name),
             partitions_def=partitions_def,
-            new_freshness_policy_by_key=_output_dict_to_asset_dict(
-                new_freshness_policy_by_output_name
-            ),
         )
 
         return AssetsDefinition.dagster_internal_init(
@@ -1728,7 +1708,6 @@ def _asset_specs_from_attr_key_params(
     descriptions_by_key: Optional[Mapping[AssetKey, str]],
     owners_by_key: Optional[Mapping[AssetKey, Sequence[str]]],
     partitions_def: Optional[PartitionsDefinition],
-    new_freshness_policy_by_key: Optional[Mapping[AssetKey, NewFreshnessPolicy]],
 ) -> Sequence[AssetSpec]:
     validated_group_names_by_key = check.opt_mapping_param(
         group_names_by_key, "group_names_by_key", key_type=AssetKey, value_type=str
@@ -1764,13 +1743,6 @@ def _asset_specs_from_attr_key_params(
         value_type=AutomationCondition,
     )
 
-    validated_new_freshness_policy_by_key = check.opt_mapping_param(
-        new_freshness_policy_by_key,
-        "new_freshness_policy_by_key",
-        key_type=AssetKey,
-        value_type=NewFreshnessPolicy,
-    )
-
     validated_owners_by_key = check.opt_mapping_param(
         owners_by_key, "owners_by_key", key_type=AssetKey, value_type=list
     )
@@ -1812,7 +1784,6 @@ def _asset_specs_from_attr_key_params(
                     partitions_def=check.opt_inst_param(
                         partitions_def, "partitions_def", PartitionsDefinition
                     ),
-                    new_freshness_policy=validated_new_freshness_policy_by_key.get(key),
                 )
             )
 

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

@@ -37,7 +37,6 @@
 from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.input import GraphIn
 from dagster._core.definitions.metadata import ArbitraryMetadataMapping, RawMetadataMapping
-from dagster._core.definitions.new_freshness_policy import NewFreshnessPolicy
 from dagster._core.definitions.output import GraphOut
 from dagster._core.definitions.partition import PartitionsDefinition
 from dagster._core.definitions.policy import RetryPolicy
@@ -91,7 +90,6 @@ def asset(
     owners: Optional[Sequence[str]] = ...,
     kinds: Optional[AbstractSet[str]] = ...,
     pool: Optional[str] = ...,
-    new_freshness_policy: Optional[NewFreshnessPolicy] = ...,
     **kwargs,
 ) -> Callable[[Callable[..., Any]], AssetsDefinition]: ...
 
@@ -120,7 +118,6 @@ def _validate_hidden_non_argument_dep_param(
 @beta_param(param="resource_defs")
 @beta_param(param="io_manager_def")
 @beta_param(param="backfill_policy")
-@beta_param(param="new_freshness_policy")
 @hidden_param(
     param="non_argument_deps",
     breaking_version="2.0.0",
@@ -170,7 +167,6 @@ def asset(
     owners: Optional[Sequence[str]] = None,
     kinds: Optional[AbstractSet[str]] = None,
     pool: Optional[str] = None,
-    new_freshness_policy: Optional[NewFreshnessPolicy] = None,
     **kwargs,
 ) -> Union[AssetsDefinition, Callable[[Callable[..., Any]], AssetsDefinition]]:
     """Create a definition for how to compute an asset.
@@ -329,7 +325,6 @@ def downstream_asset(conditional_asset):
         key=key,
         owners=owners,
         pool=pool,
-        new_freshness_policy=new_freshness_policy,
     )
 
     if compute_fn is not None:
@@ -403,7 +398,6 @@ class AssetDecoratorArgs(NamedTuple):
     check_specs: Optional[Sequence[AssetCheckSpec]]
     owners: Optional[Sequence[str]]
     pool: Optional[str]
-    new_freshness_policy: Optional[NewFreshnessPolicy]
 
 
 class ResourceRelatedState(NamedTuple):
@@ -529,7 +523,6 @@ def create_assets_def_from_fn_and_decorator_args(
             decorator_name="@asset",
             execution_type=AssetExecutionType.MATERIALIZATION,
             pool=args.pool,
-            new_freshness_policy=args.new_freshness_policy,
         )
 
         builder = DecoratorAssetsDefinitionBuilder.from_asset_outs_in_asset_centric_decorator(
@@ -709,7 +702,6 @@ def my_function(asset0):
         decorator_name="@multi_asset",
         execution_type=AssetExecutionType.MATERIALIZATION,
         pool=pool,
-        new_freshness_policy=None,
     )
 
     def inner(fn: Callable[..., Any]) -> AssetsDefinition:

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

@@ -27,7 +27,6 @@
 from dagster._core.definitions.backfill_policy import BackfillPolicy
 from dagster._core.definitions.decorators.op_decorator import _Op
 from dagster._core.definitions.input import In
-from dagster._core.definitions.new_freshness_policy import NewFreshnessPolicy
 from dagster._core.definitions.op_definition import OpDefinition
 from dagster._core.definitions.output import Out
 from dagster._core.definitions.partition import PartitionsDefinition
@@ -246,7 +245,6 @@ class DecoratorAssetsDefinitionBuilderArgs(NamedTuple):
     upstream_asset_deps: Optional[Iterable[AssetDep]]
     execution_type: Optional[AssetExecutionType]
     pool: Optional[str]
-    new_freshness_policy: Optional[NewFreshnessPolicy] = None
 
     @property
     def check_specs(self) -> Sequence[AssetCheckSpec]:
@@ -673,7 +671,6 @@ def _synthesize_specs(self) -> Sequence[AssetSpec]:
                     key,
                     deps=deps,
                     partitions_def=self.args.partitions_def,
-                    new_freshness_policy=self.args.new_freshness_policy,
                 )
             )