Thierry Hue
01/04/2023, 12:02 PM@op(out={
'start_datetime': Out(dagster_type=pd.Timestamp),
})
def calculate_build_times(context: OpExecutionContext):
job_frequency = context.op_config['job_frequency']
start_datetime = pd.Timestamp(context.run_config.get('job_start_datetime'), tz='UTC') \
if context.run_config.get('job_start_datetime') else pd.Timestamp.utcnow().floor('D')
return start_datetime
My test is:
def test_calculate_build_times():
with build_op_context(
op_config={
'job_frequency': 'DAILY',
'job_start_datetime': '2022-12-01',
}) as context:
start_datetime = calculate_build_times(context)
assert start_datetime == pd.Timestamp('2022-12-01', tz='UTC')
The test fails because context.run_config.get('job_start_datetime')
returns None
How can I set the run_config
in the context?claire
01/04/2023, 6:21 PMbuild_op_context
call. In your op, you'll need to replace every context.run_config
with context.op_config
.Thierry Hue
01/05/2023, 5:38 PMjob_start_datetime
can be overridden in the Launchpad UI when I want to run the job manually, hence the use of run_config
.
I'm trying to test different scenario with for my ops based on the configuration passed in the Launchpad.
The way I by passed the problem is to have a function get_run_config
that access the run_config, then I can mock that function to return the type of config I want but I was wondering whether there was a more elegant way to address my problem
Do you have any other idea?claire
01/05/2023, 8:23 PMbuild_op_context
and accessing config within your op via context.op_config
is the recommended pattern.
Any run config that is specified via Dagit's launchpad will be applied to the context.op_config
object.Thierry Hue
01/05/2023, 8:39 PMclaire
01/05/2023, 8:40 PMconfig_schema
field on your opThierry Hue
01/05/2023, 8:41 PM