Add dagster tags to Sling stream(s)

[laenan8466]

Documentation says, that you can add tags to sling streams by implementing a custom DagsterSlingTranslator.

I tried that:

class CustomSlingTranslator(DagsterSlingTranslator):
    @override
    def get_tags(self, stream_definition: Mapping[str, Any]) -> Mapping[str, Any]:
        config = stream_definition.get("config", {}) or {}
        meta = config.get("meta", {})
        tags = meta.get("dagster", {}).get("tags", [])
        return {tag: "" for tag in tags}

Also tried with tag: 'true' and some variations of "tag/"+tag: 'true' with no success.

The replication.yaml has the following form:

...
streams:
  my_schema.my_table:
    meta:
      dagster:
        tags: ["tag1","tag2"]

What I looked into till now:

The asset_decorators.py module from dagster_embedded_etl defines the tag loading here:

def sling_assets(...):
   ...
   return multi_asset(
    ...
    specs=[
        AssetSpec(
            ...
            tags={
                **build_kind_tag("sling"),
                **(dagster_sling_translator.get_tags(stream) or {}),
            },
            ...
        )
        for stream in streams
    ],
  )

The build kind tag seems to be correctly inserted, and when using debugging in this function, also my tags seem to match the pattern.

I guess, I'm doing something wrong. Can someone spot my flaw?

UPDATE: Was working the whole time, but I was too stupid to find the tags in the filters, see this answer.

[mlarose]

did you debug the return value of the CustomSlingTranslator.get_tags function to see if it matches your expectations?

and if it does, what you are saying is the tags does not appear on the asset in the catalog?

[laenan8466]

Yep, did some more debugging and can confirm that in the AssetSpec creation the tags list: {'dagster/kind/sling': '', 'tag1': '', 'tag2': ''}, where the first entry is the default build tag. Therefore I guess that is correct.

But your extra encouragement helped me find the error:
I expected that these tags would show up in the asset list/overview, similar to the build kind tag. But they show up only in the filters. They didn't also showed up in the runs, as this is the expected behaviour.

I changed my get_tags to the following, to allow different kind of tag definitions (including with values) in the yaml:

class CustomSlingTranslator(DagsterSlingTranslator):
    @override
    def get_tags(self, stream_definition: Mapping[str, Any]) -> Mapping[str, Any]:
        config = stream_definition.get("config", {}) or {}
        meta = config.get("meta", {})
        tags = meta.get("dagster", {}).get("tags", {})

        if isinstance(tags, list):
            return {tag: "" for tag in tags}
        if isinstance(tags, dict):
            return {tag: str(value) if value is not None else "" for tag, value in tags.items()}
        return {}

So was working the whole time along, but I didn't saw it. Would this be helpful to others, if the tags are shown with the asset?