How to add "kinds" parameter to a dbt asset?

[ssillaots-boku]

I want to add custom kinds to a dbt asset. As far as I could tell that's not possible through DagsterDbtTranslator because of

spec = AssetSpec(
...
kinds={"dbt", manifest.get("metadata", {}).get("adapter_type", "dbt")},
...

Is there some other way to do it?

I know I can add metadata but that doesn't accomplish my goal here.

The goal is to visually supplement some assets in the lineage view with custom kinds which would give a quick and comprehensive overview of some assets uniqueness/type.

[cmpadden]

Hi @ssillaots-boku , you should be able to do this in the translator, or if using Components creating a custom Component.

Create a custom translator class and override get_asset_spec() to return an AssetSpec with the kinds parameter:

from dagster_dbt import DagsterDbtTranslator
from dagster import AssetSpec

class CustomDbtTranslator(DagsterDbtTranslator):
    def get_asset_spec(self, dbt_resource_props):
        default_spec = super().get_asset_spec(dbt_resource_props)
        return default_spec._replace(kinds=["dbt", "snowflake"])  # Example

Create a custom component subclass and override get_asset_spec():

from dagster_dbt import DbtProjectComponent
from dagster import AssetSpec

class CustomDbtComponent(DbtProjectComponent):
    def get_asset_spec(self, dbt_resource_props):
        spec = super().get_asset_spec(dbt_resource_props)
        return spec._replace(kinds=["dbt", "custom_kind"])

[ssillaots-boku]

Hi @cmpadden!

First, thanks for replying!

Were you able to get the custom translator class working? I've tried quite many ways to get it working but with no luck. If you could be more precise then that would be appreciated.

[ssillaots-boku]

Was able to do it now. Doesn't look pretty but gets it done.

If you have better ways to do it then those are still welcomed.

class CustomDbtTranslator(DagsterDbtTranslator):
    def get_asset_spec(
            self,
            manifest: Mapping[str, Any],
            unique_id: str,
            project: Optional['DbtProject'],  # noqa: F821
    ) -> AssetSpec:
        default_spec = super().get_asset_spec(
            manifest=manifest,
            unique_id=unique_id,
            project=project,
        )
        updated_kinds = list(default_spec.kinds) + ['NEW_KIND']

        return AssetSpec(
            key=default_spec.key,
            deps=default_spec.deps,
            description=default_spec.description,
            metadata=default_spec.metadata,
            tags=default_spec.tags,
            group_name=default_spec.group_name,
            owners=default_spec.owners,
            kinds=updated_kinds,
        )

[ssillaots-boku]

Found a better way:

class CustomKindsDagsterDbtTranslator(CustomMetadataDagsterDbtTranslator):

    def __init__(self, custom_kinds: Optional[list[str]] = None):
        super().__init__()
        self.custom_kinds = custom_kinds

    def get_asset_spec(self, manifest: Mapping[str, Any], unique_id: str, project: Optional['DbtProject']) -> AssetSpec:
        default_spec = super().get_asset_spec(
            manifest=manifest,
            unique_id=unique_id,
            project=project,
        )

        if not self.custom_kinds:
            return default_spec

        return default_spec.merge_attributes(kinds=default_spec.kinds | set(self.custom_kinds))