I'm providing resources for source assets and normal assets.

source_assets_with_resources = with_resources(
        source_assets,
        resource_defs=resource_defs,
        resource_config_by_key=resource_config_by_key,
    )

    assets_with_resources = with_resources(
        assets,
        resource_defs=resource_defs,
        resource_config_by_key=resource_config_by_key,
    )

This code fails:

Error loading repository location repo.py:dagster._core.errors.DagsterInvalidDefinitionError: Conflicting versions of resource with key 'parquet_io_manager' were provided to different assets. When constructing a job, all resource definitions provided to assets must match by reference equality for a given key.

I'm clearly providing the same resources to different assets (

source_assets

and

assets

lists are not intersecting). This code fails in Dagster:

resource_defs_from_assets = {}
    all_assets: Sequence[Union[AssetsDefinition, SourceAsset]] = [*assets, *source_assets]
    for asset in all_assets:
        for resource_key, resource_def in asset.resource_defs.items():
            if resource_key not in resource_defs_from_assets:
                resource_defs_from_assets[resource_key] = resource_def
            if resource_defs_from_assets[resource_key] != resource_def:
                raise DagsterInvalidDefinitionError(
                    f"Conflicting versions of resource with key '{resource_key}' "
                    "were provided to different assets. When constructing a "
                    "job, all resource definitions provided to assets must "
                    "match by reference equality for a given key."
                )

so for some reason the line

if resource_defs_from_assets[resource_key] != resource_def

evaluates to True. How is this possible if I'm only defining each resource (io_manager) once? I went in with a debugger and apparently these objects are indeed different.

Hey @Daniel Gafni I’d need to dig into the source code of

with_resources

to confirm this, but I think what is happening is that within

with_resources

each resource is instantiated and attached to the assets. So when we do the comparison in the code you linked, even if we’re comparing the same resource, there are different instantiated versions of that resource for the source assets and the regular assets. I believe if you do this

assets_with_resources = with_resources(
        [*source_assets, *assets],
        resource_defs=resource_defs,
        resource_config_by_key=resource_config_by_key,
    )

it should work. If there’s another reason you need to have the source assets and regular assets given resources separately, let me know and we can figure it out

I need them separately because I work with them in a different way later on. I use the normal assets to define an asset job. For now I have a single job. The source assets just go into the repository. Is there a better way to keep different assets in separate lists?

ah i see. looping in @sandy since he may be able to give some quick advice. I’ll try to set up a small example like this and give you some recs

Thinking about it, it would be very nice if this worked: assets1, assets2, my_asset = with_resources(assets1, assets2, my_asset)

@chris @rex - I recall you did some recent work or investigation on

with_resources

and duplicate resources - do you feel equipped to answer this question?

are you providing config at any point to these resources?

Dagster's version is

1.1.5