Embedded_elt/sling: defaults are omitted in the replication config when Dagster metadata are not provided

[nixent]

What's the issue?

Embedded_elt/sling: defaults are omitted in the replication config when Dagster metadata are not provided. This is happening because of validation that checks if dagster meta is present and doesn't return defaults config.

def streams_with_default_dagster_meta(
    streams: Iterable[Mapping[str, Any]], replication_config: Mapping[str, Any]
) -> Iterable[Mapping[str, Any]]:
    """Ensures dagster meta configs in the `defaults` block of the replication_config are passed to
    the assets definition object.
    """
    default_dagster_meta = replication_config.get("defaults", {}).get("meta", {}).get("dagster", {})
    if not default_dagster_meta:
        yield from streams
    else:
        for stream in streams:
            name = stream["name"]
            config = deepcopy(stream["config"])
            if not config:
                config = {"meta": {"dagster": default_dagster_meta}}
            else:
                config["meta"] = deep_merge_dicts(
                    {"dagster": default_dagster_meta}, config.get("meta", {})
                )
            yield {"name": name, "config": config}

What did you expect to happen?

defaults from replication config should be returned regardless of dagster meta

How to reproduce?

No response

Dagster version

1.8.12

Deployment type

None

Deployment details

No response

Additional information

No response

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization. By submitting this issue, you agree to follow Dagster's Code of Conduct.

[gianfrancodemarco]

Experiencing this to, the only way seems to be to copy the object for each stream

EDIT: my problem is that default object is not passed to streams. Even if meta is present, this code only copies that key, so I don't know If this is the same problem