I've tried to use `with_resources` to create multi...
# ask-community
d
I've tried to use
with_resources
to create multiple assets from the same graph with varying configs, using the code below
Copy code
@op(required_resource_keys={"a"})
def op1(context):
    return context.resources.a["v"] + 1


@op
def op2(context, op1):
    return op1 + 1


@graph
def mygraph():
    return op2(op1())


@job(resource_defs={"a": make_values_resource(v=int)})
def myjob():
    mygraph()


VALUE_OPTIONS = [0, 1, 2]
myassets = [
    with_resources(
        [AssetsDefinition.from_graph(mygraph)],
        resource_defs={"a": make_values_resource(v=int)},
        resource_config_by_key={"a": {"config": {"v": v}}},
    )[0]
    for v in VALUE_OPTIONS
]


@repository
def test_repo():
    return [*myassets, myjob]
However, I get the following error when I attempt to load it
Copy code
dagster.core.errors.DagsterInvalidDefinitionError: Conflicting versions of resource with key 'a' were provided to different assets. When constructing a job, all resource definitions provided to assets must match by reference equality for a given key.
Is it not possible to specify multiple assets from a single graph that vary only in input config? I realise this a classic case for Static Partitioning, but I'm trying to do this without using Static Partitioning to get around the fact that I can't partition across multiple directions, as mentioned in a previous thread.
🤖 1
s
Hi @Dmitry Mogilevsky - resources are typically used when you want to supply the same object or configuration to different ops or assets. If you just want vary configuration, you could try something like this:
Copy code
from dagster import op, graph, AssetsDefinition, repository, config_mapping


@op(config_schema={"v": int})
def op1(context):
    return context.op_config["v"] + 1


@op
def op2(context, op1):
    return op1 + 1


@config_mapping(config_schema={"v": int})
def mygraph_config_mapping(val):
    return {"op1": {"config": val}}


@graph(config=mygraph_config_mapping)
def mygraph():
    return op2(op1())


VALUE_OPTIONS = [0, 1, 2]
myassets = [
    [AssetsDefinition.from_graph(mygraph.configured({"v": v}, name=f"mygraph{v}"))][0]
    for v in VALUE_OPTIONS
]


@repository
def test_repo():
    return [*myassets]
d
I want to supply the same the same configuration to multiple ops in a single instance of a graph, but I want to do it for multiple instances of the same graph, each with different configuration set, if that makes sense. But this is fine, that's a problem I've solved before (by converting configs into inputs that then go into all the dependent ops). The only knowledge I didn't have for this solution is the .configured function of the graph object. This should be enough to sort me out, thank you!
👍 1