Matthew Heguy
03/10/2023, 3:29 PM@op(config_schema={"header": bool})
def get_header(context) -> str:
return context.op_config["header"]
@asset(partitions_def=parts)
def asset_down(asset_up: str) -> None:
print(get_header())
print(asset_up)
I've tried a few different approaches but haven't figured it out..
if __name__ == "__main__":
context = build_op_context(op_config={"header": "Hello"})
try:
asset_down(context, asset_up="1")
except DagsterInvalidInvocationError:
"Compute function of op 'asset_down' has no context argument, but context was provided when invoking."
try:
asset_down(asset_up="1")
except DagsterInvalidInvocationError:
"Compute function of op 'get_header' has context argument, but no context was provided when invoking."
try:
asset_down.execute_in_process(context)
except AttributeError:
"'AssetsDefinition' object has no attribute 'execute_in_process'"
Zach
03/10/2023, 4:55 PMops
from within an asset
, they should be called from a `graph`/`job`. However, you can define a config_schema on your assets if that gets you closer to what you're trying to do:
@asset(partitions_def=parts, config_schema={"header": bool})
def asset_down(asset_up: str) -> None:
print(get_header())
print(asset_up)
Matthew Heguy
03/10/2023, 4:56 PMZach
03/10/2023, 4:58 PMMatthew Heguy
03/10/2023, 4:59 PM