Chris Histe
09/28/2022, 2:01 PMContext
]
Hello there,
is there an equivalent of build_init_resource_context
for StepExecutionContext
. I’m unit testing a custom IO Manager and getting an error when accessing _context_.has_partition_key
(InputContext). It seems you can pass a StepExecutionContext
to build_input_context
but I’m not sure how to instantiate that class.dagster._core.errors.DagsterInvariantViolationError: Attempting to access step_context, but it was not provided when constructing the InputContext
The errorStepExecutionContext
so I assume there must a helper function for test.claire
09/28/2022, 11:25 PMStepExecutionContext
. The limitation here is that some information, including the partition key, is only available on the pipeline run object (and this information is then passed to the StepInputContext
). Because a input context constructed via build_input_context
is not associated with a pipeline run, these fields cause an error.partitions_def = StaticPartitionsDefinition(["a", "b", "c", "d"])
class MyIOManager(IOManager):
def handle_output(self, context, obj):
assert context.asset_partition_key == "b"
def load_input(self, context):
...
@asset(partitions_def=partitions_def)
def my_asset(context):
assert context.asset_partitions_def_for_output() == partitions_def
my_job = build_assets_job(
"my_job",
assets=[my_asset],
resource_defs={"io_manager": IOManagerDefinition.hardcoded_io_manager(MyIOManager())},
)
result = my_job.execute_in_process(partition_key="b")
Chris Histe
09/28/2022, 11:38 PMclaire
09/28/2022, 11:39 PM@asset
in each unit test, or you could share the same @asset
/ job across all unit tests and provide different partition keys, depending your needsChris Histe
09/29/2022, 3:37 PM_context_.has_partition_key
?claire
09/29/2022, 7:18 PMbuild_input_context
. Maybe I can put together a PR quickly to allow for the same functionality on build_input_context
🙂Chris Histe
09/29/2022, 7:45 PMbuild_init_resource_context
. I have dagster_run_id = _context_.run_id
in my code and it’s causing some errors because it’s None
during my unit test
So as a workaround I’m doing the folllowing but it’s also a smell
if "PYTEST_CURRENT_TEST" in os.environ:
dagster_run_id = "unit-testing"
else:
dagster_run_id = context.run_id
claire
09/29/2022, 7:51 PMChris Histe
09/29/2022, 7:53 PMclaire
09/29/2022, 8:03 PMChris Histe
09/29/2022, 8:06 PM