[freshness] rename freshness_policy property, arg name to legacy_freshness_policy (dagster-io#30771)

## Summary & Motivation
Allows us to use `freshness_policy` as the property or argument name for the new `FreshnessPolicy`.


## How I Tested These Changes
unit tests

## Changelog

> Insert changelog entry or delete this section.

Author: OwenKephart

examples/starlift-demo/dbt_example/dagster_defs/utils.py

@@ -98,7 +98,7 @@ def with_deps(
         group_names_by_key=assets_def.group_names_by_key,
         metadata_by_key=assets_def.metadata_by_key,
         tags_by_key=assets_def.tags_by_key,
-        freshness_policies_by_key=assets_def.freshness_policies_by_key,
+        legacy_freshness_policies_by_key=assets_def.legacy_freshness_policies_by_key,
         backfill_policy=assets_def.backfill_policy,
         descriptions_by_key=assets_def.descriptions_by_key,
         check_specs_by_output_name=assets_def.check_specs_by_output_name,

python_modules/dagster-graphql/dagster_graphql/schema/asset_graph.py

@@ -833,7 +833,7 @@ def resolve_dependencies(self, graphene_info: ResolveInfo) -> Sequence[GrapheneA
     def resolve_freshnessInfo(
         self, graphene_info: ResolveInfo
     ) -> Optional[GrapheneAssetFreshnessInfo]:
-        if self._asset_node_snap.freshness_policy:
+        if self._asset_node_snap.legacy_freshness_policy:
             return get_freshness_info(
                 asset_key=self._asset_node_snap.asset_key,
                 data_time_resolver=graphene_info.context.data_time_resolver,
@@ -843,8 +843,8 @@ def resolve_freshnessInfo(
     def resolve_freshnessPolicy(
         self, _graphene_info: ResolveInfo
     ) -> Optional[GrapheneFreshnessPolicy]:
-        if self._asset_node_snap.freshness_policy:
-            return GrapheneFreshnessPolicy(self._asset_node_snap.freshness_policy)
+        if self._asset_node_snap.legacy_freshness_policy:
+            return GrapheneFreshnessPolicy(self._asset_node_snap.legacy_freshness_policy)
         return None
 
     def resolve_internalFreshnessPolicy(

python_modules/dagster-graphql/dagster_graphql_tests/graphql/repo.py

@@ -1832,7 +1832,7 @@ def fresh_diamond_right(fresh_diamond_top):
 
 
 @asset(
-    freshness_policy=LegacyFreshnessPolicy(maximum_lag_minutes=30),
+    legacy_freshness_policy=LegacyFreshnessPolicy(maximum_lag_minutes=30),
     auto_materialize_policy=AutoMaterializePolicy.lazy(),
 )
 def fresh_diamond_bottom(fresh_diamond_left, fresh_diamond_right):

python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_freshness.py

@@ -29,7 +29,7 @@
 """
 
 
-@observable_source_asset(freshness_policy=LegacyFreshnessPolicy(maximum_lag_minutes=30))
+@observable_source_asset(legacy_freshness_policy=LegacyFreshnessPolicy(maximum_lag_minutes=30))
 def source_asset_with_freshness(context):
     return ObserveResult(metadata={DATA_TIME_METADATA_KEY: float(context.run.tags["data_time"])})
 

python_modules/dagster-test/dagster_test/toys/auto_materializing/lazy_repo_2.py

@@ -25,7 +25,7 @@ def lazy_downstream_3(lazy_downstream_1):
 
 @asset(
     auto_materialize_policy=AutoMaterializePolicy.lazy(),
-    freshness_policy=LegacyFreshnessPolicy(maximum_lag_minutes=1440),
+    legacy_freshness_policy=LegacyFreshnessPolicy(maximum_lag_minutes=1440),
 )
 def lazy_downstream_4(lazy_downstream_2, lazy_downstream_3):
     return lazy_downstream_2 + lazy_downstream_3
@@ -52,7 +52,7 @@ def lazy_downstream_3_partitioned(lazy_downstream_1_partitioned):
 @asset(
     partitions_def=daily_partitions_def,
     auto_materialize_policy=AutoMaterializePolicy.lazy(),
-    freshness_policy=LegacyFreshnessPolicy(maximum_lag_minutes=1440),
+    legacy_freshness_policy=LegacyFreshnessPolicy(maximum_lag_minutes=1440),
 )
 def lazy_downstream_4_partitioned(lazy_downstream_2_partitioned, lazy_downstream_3_partitioned):
     return lazy_downstream_2_partitioned + lazy_downstream_3_partitioned

python_modules/dagster/dagster/_core/definitions/asset_checks.py

@@ -37,7 +37,7 @@ def create(
             group_names_by_key=None,
             metadata_by_key=None,
             tags_by_key=None,
-            freshness_policies_by_key=None,
+            legacy_freshness_policies_by_key=None,
             backfill_policy=None,
             descriptions_by_key=None,
             check_specs_by_output_name=check_specs_by_output_name,

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

@@ -113,8 +113,8 @@ def partition_mappings(self) -> Mapping[AssetKey, PartitionMapping]:  # pyright:
         return self._spec.partition_mappings
 
     @property
-    def freshness_policy(self) -> Optional[LegacyFreshnessPolicy]:
-        return self._spec.freshness_policy
+    def legacy_freshness_policy(self) -> Optional[LegacyFreshnessPolicy]:
+        return self._spec.legacy_freshness_policy
 
     @property
     def auto_materialize_policy(self) -> Optional[AutoMaterializePolicy]:

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

@@ -2,7 +2,12 @@
 from typing import Any, Optional, Union
 
 import dagster._check as check
-from dagster._annotations import hidden_param, only_allow_hidden_params_in_kwargs, public
+from dagster._annotations import (
+    deprecated_param,
+    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
@@ -31,11 +36,7 @@
 EMPTY_ASSET_KEY_SENTINEL = AssetKey([])
 
 
-@hidden_param(
-    param="freshness_policy",
-    breaking_version="1.10.0",
-    additional_warn_text="use freshness checks instead",
-)
+@deprecated_param(param="legacy_freshness_policy", breaking_version="1.12.0")
 @hidden_param(
     param="auto_materialize_policy",
     breaking_version="1.10.0",
@@ -65,8 +66,6 @@ class AssetOut:
         group_name (Optional[str]): A string name used to organize multiple assets into groups. If
             not provided, the name "default" is used.
         code_version (Optional[str]): The version of the code that generates this asset.
-        freshness_policy (Optional[FreshnessPolicy]): (Deprecated) A policy which indicates how up
-            to date this asset is intended to be.
         automation_condition (Optional[AutomationCondition]): AutomationCondition to apply to the
             specified asset.
         backfill_policy (Optional[BackfillPolicy]): BackfillPolicy to apply to the specified asset.
@@ -102,6 +101,7 @@ def __init__(
         owners: Optional[Sequence[str]] = None,
         tags: Optional[Mapping[str, str]] = None,
         kinds: Optional[set[str]] = None,
+        legacy_freshness_policy: Optional[LegacyFreshnessPolicy] = None,
         **kwargs,
     ):
         # Accept a hidden "spec" argument to allow for the AssetOut to be constructed from an AssetSpec
@@ -119,7 +119,6 @@ def __init__(
         )
 
         auto_materialize_policy = kwargs.get("auto_materialize_policy")
-        freshness_policy = kwargs.get("freshness_policy")
         has_any_spec_args = any(
             [
                 key,
@@ -129,7 +128,7 @@ def __init__(
                 code_version,
                 automation_condition,
                 auto_materialize_policy,
-                freshness_policy,
+                legacy_freshness_policy,
                 owners,
                 tags,
                 kinds,
@@ -156,8 +155,10 @@ def __init__(
                     "automation_condition",
                     AutomationCondition,
                 ),
-                freshness_policy=check.opt_inst_param(
-                    freshness_policy, "freshness_policy", LegacyFreshnessPolicy
+                legacy_freshness_policy=check.opt_inst_param(
+                    legacy_freshness_policy,
+                    "legacy_freshness_policy",
+                    LegacyFreshnessPolicy,
                 ),
                 owners=check.opt_sequence_param(owners, "owners", of_type=str),
                 tags=normalize_tags(tags or {}, strict=True),
@@ -190,8 +191,8 @@ def code_version(self) -> Optional[str]:
         return self._spec.code_version
 
     @property
-    def freshness_policy(self) -> Optional[LegacyFreshnessPolicy]:
-        return self._spec.freshness_policy
+    def legacy_freshness_policy(self) -> Optional[LegacyFreshnessPolicy]:
+        return self._spec.legacy_freshness_policy
 
     @property
     def automation_condition(self) -> Optional[AutomationCondition]:

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

@@ -104,9 +104,9 @@ def validate_kind_tags(kinds: Optional[AbstractSet[str]]) -> None:
 
 
 @hidden_param(
-    param="freshness_policy",
-    breaking_version="1.10.0",
-    additional_warn_text="use freshness checks instead.",
+    param="legacy_freshness_policy",
+    breaking_version="1.12.0",
+    additional_warn_text="use the freshness policy abstraction instead.",
 )
 @hidden_param(
     param="auto_materialize_policy",
@@ -161,7 +161,7 @@ class AssetSpec(IHasInternalInit, IHaveNew, LegacyNamedTupleMixin):
     group_name: PublicAttr[Optional[str]]
     skippable: PublicAttr[bool]
     code_version: PublicAttr[Optional[str]]
-    freshness_policy: PublicAttr[Optional[LegacyFreshnessPolicy]]
+    legacy_freshness_policy: PublicAttr[Optional[LegacyFreshnessPolicy]]
     automation_condition: PublicAttr[Optional[AutomationCondition]]
     owners: PublicAttr[Sequence[str]]
     tags: PublicAttr[Mapping[str, str]]
@@ -225,9 +225,9 @@ def __new__(
             skippable=check.bool_param(skippable, "skippable"),
             group_name=check.opt_str_param(group_name, "group_name"),
             code_version=check.opt_str_param(code_version, "code_version"),
-            freshness_policy=check.opt_inst_param(
-                kwargs.get("freshness_policy"),
-                "freshness_policy",
+            legacy_freshness_policy=check.opt_inst_param(
+                kwargs.get("legacy_freshness_policy"),
+                "legacy_freshness_policy",
                 LegacyFreshnessPolicy,
             ),
             automation_condition=check.opt_inst_param(
@@ -270,7 +270,7 @@ def dagster_internal_init(
             skippable=skippable,
             group_name=group_name,
             code_version=code_version,
-            freshness_policy=kwargs.get("freshness_policy"),
+            legacy_freshness_policy=kwargs.get("legacy_freshness_policy"),
             automation_condition=automation_condition,
             owners=owners,
             tags=tags,
@@ -330,7 +330,7 @@ def replace_attributes(
         tags: Optional[Mapping[str, str]] = ...,
         kinds: Optional[set[str]] = ...,
         partitions_def: Optional[PartitionsDefinition] = ...,
-        freshness_policy: Optional[LegacyFreshnessPolicy] = ...,
+        legacy_freshness_policy: Optional[LegacyFreshnessPolicy] = ...,
     ) -> "AssetSpec":
         """Returns a new AssetSpec with the specified attributes replaced."""
         current_tags_without_kinds = {
@@ -347,7 +347,7 @@ def replace_attributes(
                 skippable=skippable if skippable is not ... else self.skippable,
                 group_name=group_name if group_name is not ... else self.group_name,
                 code_version=code_version if code_version is not ... else self.code_version,
-                freshness_policy=self.freshness_policy,
+                legacy_freshness_policy=self.legacy_freshness_policy,
                 automation_condition=automation_condition
                 if automation_condition is not ...
                 else self.automation_condition,
@@ -396,7 +396,7 @@ def merge_attributes(
                 skippable=self.skippable,
                 group_name=self.group_name,
                 code_version=self.code_version,
-                freshness_policy=self.freshness_policy,
+                legacy_freshness_policy=self.legacy_freshness_policy,
                 automation_condition=self.automation_condition,
                 owners=[*self.owners, *(owners if owners is not ... else [])],
                 tags={**current_tags_without_kinds, **(tags if tags is not ... else {})},