Josh Lloyd
07/21/2021, 6:27 PM@pipeline(
mode_defs=[
mode_def_prod,
mode_def_prod_coll,
],
)
def collective_pipeline():
...
@schedule(
pipeline_name="collective_pipeline",
cron_schedule="0 7 * * *",
execution_timezone="UTC",
mode='prod_coll'
)
def coll_daily_schedule():
return {"resources": {"io_manager": {"config": {"s3_bucket": "dagster-wdt", "s3_prefix": "stage/io"}}}}
def test_coll_daily_schedule():
run_config = coll_daily_schedule()
assert validate_run_config(collective_pipeline, run_config)
error when I use pytest
in commandline:
dagster.core.errors.DagsterInvalidConfigError: Error in config for pipeline collective_pipeline
Error 1: Received unexpected config entries "['s3_bucket', 's3_prefix']" at path root:resources:io_manager:config. Expected: "['base_dir']."
One strange solution I have found is to switch the order of the mode definitions in the pipeline so that mode_def_prod_coll
comes first. Then the test passes. But in my mind, this shouldn’t happen because I’m specifically calling out the name of the mode I want to use in the decorator of the schedule prod_coll
alex
07/21/2021, 6:52 PMvalidate_run_config
takes a mode
argument, which you could fill out with mode=call_daily_schedule.mode
Josh Lloyd
07/21/2021, 6:53 PM