How to configure FreshnessPolicy for dbt assets after Dagster 1.12.0

Last updated: November 21, 2025

Problem Description

  Starting with Dagster 1.12.0, the get_freshness_policy method was removed from DagsterDbtTranslator, breaking existing configurations that relied on this method to set freshness policies for dbt assets. Users need a new approach to configure freshness policies without selecting assets individually.

Symptoms

  • Missing get_freshness_policy method in DagsterDbtTranslator

  • Legacy LegacyFreshnessPolicy objects are no longer automatically parsed from dbt model configuration

  • Existing freshness policy configurations stop working after upgrading to Dagster 1.12.0

Root Cause

  Dagster 1.12.0 removed the get_freshness_policy method from DagsterDbtTranslator as part of deprecating legacy freshness policy handling. The new approach requires using the get_asset_spec method to configure freshness policies.

Solution

Quick Fix (if applicable)

  Use dbt meta configuration with +meta.dagster.freshness_policy in your dbt project files for simple cases.

Step-by-Step Resolution

  1. Override the get_asset_spec method in your custom DagsterDbtTranslator class:

def get_asset_spec(
    self,
    manifest: Mapping[str, Any],
    unique_id: str,
    project: Optional[Any] = None,
) -> dg.AssetSpec:
    # Get the existing asset_spec from the parent class to preserve all existing config
    default_asset_spec = super().get_asset_spec(manifest, unique_id, project)
    
    # Get dbt_resource_props to access metadata
    dbt_resource_props = self.get_resource_props(manifest, unique_id)
    dagster_metadata = dbt_resource_props.get("meta", {}).get("dagster", {})
    freshness_policy_config = dagster_metadata.get("freshness_policy", {})
    type = freshness_policy_config.get("type")

    if type == "cron":
        deadline_cron = freshness_policy_config.get("deadline_cron")
        lower_bound_delta = freshness_policy_config.get("lower_bound_delta")
        timezone = freshness_policy_config.get("timezone", "Etc/UTC")

        asset_spec = default_asset_spec.replace_attributes( 
            freshness_policy=dg.FreshnessPolicy.cron(
                deadline_cron=deadline_cron,
                lower_bound_delta=timedelta(minutes=lower_bound_delta),
                timezone=timezone,
            ),
        )
    elif type == "time_window":
        fail_window = freshness_policy_config.get("fail_window")
        warn_window = freshness_policy_config.get("warn_window")

        asset_spec = default_asset_spec.replace_attributes(
            freshness_policy=dg.FreshnessPolicy.time_window(
                fail_window=timedelta(minutes=fail_window),
                warn_window=timedelta(minutes=warn_window) if warn_window is not None else None,
            ),
        )
        
    else:
        asset_spec = default_asset_spec
        
    return asset_spec

  1. Configure freshness policies in your dbt model files using meta configuration:

# In your dbt schema.yml or model files
models:
  - name: users
    config:
      meta:
        dagster:
          freshness_policy:
            type: "cron"
            deadline_cron: "0 9 * * *"
            lower_bound_delta: 60
            timezone: "Etc/UTC"

  1. Verify the configuration is working by checking that your assets now have the expected freshness policies applied

Alternative Solutions (if applicable)

  For simple cases, you can use the built-in dbt meta configuration without custom translator code by setting +meta.dagster.freshness_polic directly in your dbt project configuration.

Prevention

  When upgrading Dagster versions, review the release notes for breaking changes to core APIs like DagsterDbtTranslator. Test freshness policy configurations in a development environment before deploying to production.

Related Documentation