Heya, I'm a bit stuck trying to test a partitioned asset that has an op that requires a config. I feel like maybe I'm just not designing things properly? Short example of what I mean:

@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'"

You can't call

ops

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)

Ok, so that's my underlying issue: I should be creating a graph-backed asset in this case?

Yeah it just depends on what you're trying to do. If you want to have pieces of your asset be composable and observable as separate steps then yes, you'll probably want to head in the graph-backed asset direction. If the logic composing your asset doesn't benefit from being broken up you can just put it in the asset definition.

Perfect, thank you. 🙂