Defining which preset to use for a schedule

[blarghmatey]

In defining a schedule, is there a way to specify which preset to use for a given pipeline? I would like to execute the same pipeline code using 2 different presets to load separate configurations for different target sources on a daily schedule for each.

[helloworld]

Hi @blarghmatey,

You can construct a schedule definition using a preset like this:

@daily_schedule(
    start_date=...,
    pipeline_name="my_pipeline",
    solid_selection=preset.solid_selection,
    mode=preset.mode,
    tags_fn_for_date=lambda _: preset.tags,
)
def my_daily_schedule(_date):
    return preset.run_config

[blarghmatey]

So, in that scenario should I define the presets as separate objects so that they can be importable or used directly in the schedule decorator, as opposed to be defined inline in the pipeline decorator as is shown in some of the examples?

e.g. rather than doing

@pipeline(
    preset_defs=[
        PresetDefinition.from_files(
            name='residential',
            config_files=[
                '/etc/dagster/residential_edx.yaml'
            ],
            mode='production',
            tags={
                'business_unit': 'residential'
            }
        ),
        PresetDefinition.from_files(
            name='xpro',
            config_files=[
                '/etc/dagster/xpro_edx.yaml'
            ],
            mode='production',
            tags={
                'business_unit': 'mitxpro'
            }
        )
    ]
)
def foobar()

I would do

xpro_def = PresetDefinition.from_files(
    name='xpro',
    config_files=[
        '/etc/dagster/xpro_edx.yaml'
    ],
    mode='production',
    tags={'business_unit': 'mitxpro'})

@pipeline(preset_defs=[xpro_def])
...

[blarghmatey]

For a more concrete reference, this is the code I am working from https://github.com/mitodl/ol-data-pipelines/blob/main/src/ol_data_pipelines/edx/solids.py#L556-L607

[blarghmatey]

After looking through the code, it looks like I can just import the pipeline object and use the get_preset('xpro') method rather than declaring the presets at the module level.

[helloworld]

Yes exactly. You need to define the preset as a separate object or use the get_preset method.

If you find yourself doing this often, you could write a helper:

from dagster import check, PipelineDefinition, DagsterInvalidDefinitionError, daily_schedule
import datetime


def daily_schedule_definition_from_pipeline_preset(pipeline, preset_name, start_date):
    check.inst_param(pipeline, "pipeline", PipelineDefinition)
    check.str_param(preset_name, "preset_name")
    check.inst_param(start_date, "start_date", datetime.datetime)

    preset = pipeline.get_preset(preset_name)
    if not preset:
        raise DagsterInvalidDefinitionError(
            "Preset {preset_name} was not found "
            "on pipeline {pipeline_name}".format(
                preset_name=preset_name, pipeline_name=pipeline.name
            )
        )

    @daily_schedule(
        start_date=start_date,
        pipeline_name=pipeline.name,
        solid_selection=preset.solid_selection,
        mode=preset.mode,
        tags_fn_for_date=lambda _: preset.tags,
    )
    def my_daily_schedule(_date):
        return preset.run_config

    return my_daily_schedule