How to use a custom resource key when using @dbt_assets

[brandybobandyistaken]

Per documentation here: https://docs.dagster.io/_apidocs/libraries/dagster-dbt#dagster_dbt.dbt_assets

The @dbt_assets decorator should be able to customize the dbt_resource_key, which was deprecated from the dagster_dbt.load_assets_from_dbt_project and dagster_dbt.load_assets_from_dbt_manifest functions.

However, I don't see a way to do this. This is a problem, because there is no way to choose which DbtCliResource to assign to a group of dbt assets. This means you cannot govern multiple dbt projects within one instance of dagster.

[rexledesma]

In our API examples, we use Pythonic Dagster Resources. There, the parameter name of the resource is associated with the resource key in the Definitions object. Here's some relevant documentation.

So, you can just customize the parameter name if you want to use a different resource key, similar to customizing dbt_resource_key in load_assets_from_dbt_project or load_assets_from_dbt_manifest. Here's an example, with the relevant lines highlighted.

@dbt_assets(manifest=dbt_manifest_path)
def jaffle_shop_dbt_assets(
+   context: AssetExecutionContext, my_custom_dbt_resource_key: DbtCliResource
):
    yield from my_custom_dbt_resource_key.cli(["build"], context=context).stream()
    
defs = Definitions(
    assets=[jaffle_shop_dbt_assets],
    resources={
+       "my_custom_dbt_resource_key": DbtCliResource(...),
    },
)

If you're dynamically generating your dbt assets, then you can also just dynamically retrieve the resource associated with your dbt project. Here's another example:

dbt_projects = [
    {
        "project_dir": Path("my/path/to/project/a"),
        "dbt_manifest_path": Path("my/path/to/project/a/manifest"),
        "name": "dbt_project_a"
    },
    {
        "project_dir": Path("my/path/to/project/b"),
        "dbt_manifest_path": Path("my/path/to/project/a/manifest"),
        "name": "dbt_project_b"
    },
]


def create_my_dbt_assets_from_dbt_project(dbt_project: dict) -> AssetsDefinition:
    dbt_project_name = dbt_project["name"]

    @dbt_assets(
        name=dbt_project_name,
        manifest=dbt_project["dbt_manifest_path"],
        required_resource_keys={dbt_project_name},
    )
    def jaffle_shop_dbt_assets(context: AssetExecutionContext):
        dbt = getattr(context.resources, dbt_project_name)
        yield from dbt.cli(["build"], context=context).stream()
    
    return jaffle_shop_dbt_assets


defs = Definitions(
    assets=[
        create_my_dbt_assets_from_dbt_project(dbt_project)
        for dbt_project in dbt_projects
    ],
    resources={
       dbt_project["name"]: DbtCliResource(project_dir=dbt_project["project_dir"]
       for dbt_project in dbt_projects
    },
)

[brandybobandyistaken]

This worked thank you very much!

I suppose I still don't understand the link between

1. the parameter name in the function @dbt_assets is decorating

and

2. the resource key provided to definitions.

How does the asset know to look for a resource with the same name as the parameter?

[ldnicolasmay]

Using Dagster v1.8.4 here.

Following this solution exactly, I kept getting a dagster._core.errors.DagsterUnknownResourceError:

dagster._core.errors.DagsterUnknownResourceError: Unknown resource `dbt_project_a`. Specify `dbt_project_a` as a required resource on the compute / config function that accessed it.

What fixed it for me was adding the required_resource_keys parameter to the @dbt_assets decorator:

dbt_projects = [
    {
        "project_dir": Path("my/path/to/project/a"),
        "dbt_manifest_path": Path("my/path/to/project/a/manifest"),
        "name": "dbt_project_a"
    },
    {
        "project_dir": Path("my/path/to/project/b"),
        "dbt_manifest_path": Path("my/path/to/project/b/manifest"),
        "name": "dbt_project_b"
    },
]


def create_my_dbt_assets_from_dbt_project(dbt_project: dict) -> AssetsDefinition:
    dbt_project_name = dbt_project["name"]

    @dbt_assets(
        name=dbt_project_name,
        manifest=dbt_project["dbt_manifest_path"],
        required_resource_keys={dbt_project_name},  # <======= CHANGE RIGHT HERE <========
    )
    def jaffle_shop_dbt_assets(context: AssetExecutionContext):
        dbt = getattr(context.resources, dbt_project_name)
        yield from dbt.cli(["build"], context=context).stream()
    
    return jaffle_shop_dbt_assets


defs = Definitions(
    assets=[
        create_my_dbt_assets_from_dbt_project(dbt_project)
        for dbt_project in dbt_projects
    ],
    resources={
       dbt_project["name"]: DbtCliResource(project_dir=dbt_project["project_dir"]
       for dbt_project in dbt_projects
    },
)

[rexledesma]

Thanks @ldnicolasmay, I updated the example to use required_resource_keys.