def METHOD_NAME(popularity_refresh: PopularityRefresh):
    """
    Instantiate a DAG for a popularity refresh.
    For the given media type, this DAG will recalculate the popularity constants and
    then refresh all existing standardized popularity scores with the new constant.
    Required Arguments:
    popularity_refresh: dataclass containing configuration information for the DAG
    """
    default_args = {
        **DAG_DEFAULT_ARGS,
        **popularity_refresh.default_args,
    }
    dag = DAG(
        dag_id=popularity_refresh.dag_id,
        default_args=default_args,
        start_date=popularity_refresh.start_date,
        schedule=popularity_refresh.schedule,
        max_active_runs=1,
        catchup=False,
        doc_md=__doc__,
        tags=["popularity_refresh"],
    )
    with dag:
        # Refresh the underlying popularity tables. This step recalculates the
        # popularity constants, which will later be used to calculate updated
        # standardized popularity scores.
        refresh_popularity_metrics = create_refresh_popularity_metrics_task_group(
            popularity_refresh
        )
        # Once popularity constants have been calculated, establish the cutoff time
        # after which records that have updates do not need to be refreshed.
        get_cutoff_time = get_last_updated_time()
        # For each provider that supports popularity data for this media type, trigger a
        # batched_update to recalculate all old standardized popularity scores using the
        # newly refreshed constant.
        refresh_popularity_scores = TriggerDagRunOperator.partial(
            task_id="refresh_popularity",
            trigger_dag_id=BATCHED_UPDATE_DAG_ID,
            # Wait for all the dagruns to finish
            wait_for_completion=True,
            # Release the worker slot while waiting
            deferrable=True,
            poke_interval=popularity_refresh.poke_interval,
            execution_timeout=popularity_refresh.refresh_popularity_timeout,
            retries=0,
        ).expand(
            # Build the conf for each provider
            conf=get_providers_update_confs(
                POSTGRES_CONN_ID, popularity_refresh, get_cutoff_time
            )
        )
        notify_complete = notify_slack(
            text=(
                f"{popularity_refresh.media_type.capitalize()} Popularity Refresh"
                " Complete"
            ),
            media_type=popularity_refresh.media_type,
            dag_id=popularity_refresh.dag_id,
        )
        # Set up task dependencies
        (
            refresh_popularity_metrics
            >> get_cutoff_time
            >> refresh_popularity_scores
            >> notify_complete
        )
    return dag