Scheduling Time-Partitioned Asset Advice

[patrickautone]

I am using Dagster to run dbt models. I have an asset that I have partitioned based on Week (starting Monday):

import json

from dagster import (
    AssetExecutionContext,
    BackfillPolicy,
    WeeklyPartitionsDefinition,
)
from dagster_dbt import DbtCliResource, dbt_assets

WEEK_PARTITION = WeeklyPartitionsDefinition(
    start_date="2024-05-01", day_offset=1, end_offset=1
)

@dbt_assets(
    manifest=<manifest_path>,
    partitions_def=WEEK_PARTITION,
    project=<dbt_project>,
    backfill_policy=BackfillPolicy.single_run(),
    select="group:dbt_mart",
)
def dbt_model_assets_by_date(
    context: AssetExecutionContext,
    dbt: DbtCliResource,
):
    # Fetch the partition date from context
    partition_date = context.partition_key

    dbt_vars = {
        "partition_date": partition_date,
        "dagster_run_id": context.run_id,
    }

    dbt_build_args = [
        "build",
        "--vars",
        json.dumps(dbt_vars),
    ]

    yield from (
        dbt.cli(dbt_build_args, context=context)
        .stream()
        .fetch_column_metadata(with_column_lineage=True)
    )

and a job that runs that asset:

from dagster import define_asset_job
from dagster_dbt import build_dbt_asset_selection
from dataflow.assets.assets import dbt_model_assets_by_date

dbt_job = define_asset_job(
    "dbt_job",
    selection=build_dbt_asset_selection(
        [dbt_model_assets_by_date], dbt_select="group:dbt_mart"
    ),
)

As the docs suggest I have tried to create a schedule using build_schedule_from_partitioned_job so that I can run the job every Tuesday (note the difference in schedule from the partition, despite having the same cadence). I attempted the below code and I get the following error.

from dagster import build_schedule_from_partitioned_job, DefaultScheduleStatus

job_schedule = build_schedule_from_partitioned_job(
    dbt_job,
    description='Runs every Tuesday at 3:25 PM based on weekly partition.',
    cron_schedule="25 15 * * 2",  # This cron runs every Tuesday at 15:25
    default_status=DefaultScheduleStatus.RUNNING,
)

Error:

dagster._check.functions.CheckError: Failure condition: Cannot provide cron_schedule or execution_timezone to build_schedule_from_partitioned_job for a time-partitioned job.

Any recommendations? I'm pretty stuck. The docs talk about how you can schedule jobs with time based partitions and I can't work out if I've misnunderstood, if I've written the code wrong or if it's a bug.

[nixent]

Your JobDefinition variable is dbt_build_forecast_accuracy, however you are passing dbt_job.

Can you try passing dbt_build_forecast_accuracy?

job_schedule = build_schedule_from_partitioned_job(
    dbt_build_forecast_accuracy,
   ...

[patrickautone]

Apologies I copied my code over incorrectly, I've updated it now. The problem still stands sadly

[patrickautone]

This was resolved by using "day_of_week" rather than cron_schedule. Understandably, it seems these non-cron parameters can only be used if they "make sense" with the partition cadence.