Hi, I am using

dagster-dbt

integration, and I want to determine the

project_dir

dynamically. So I came up with

from dagster import file_relative_path, resource
from dagster_dbt import dbt_cli_resource

@resource(required_resource_keys={"input_config"})
def dbt_resource(init_context: InitResourceContext):
    company_name = init_context.resources.input_config["company_name"].lower()
    dbt_project_dir = file_relative_path(__file__, f"./events/{company_name}")
    return dbt_cli_resource.configured(
            {"project_dir": dbt_project_dir, "profiles_dir": dbt_project_dir}
        )

where

input_config

is another resource. Then, I add it in

resource_defs

of a

ModeDefinition

with

"dbt": dbt_resource

. Now, in solids where I want to use this, I have

required_resource_keys={"input_config", "dbt"}

in their

@solid

decorator. Now, I want to access

context.resources.dbt.cli()

inside that solid. The issue is that it should have been an object of

dagster_dbt.cli.resources.DbtCliResource

but unfortunately, it’s of type

ResourceDefinition

due to which there is no

.cli()

method available to it. How can I ensure that it is of correct type aka ``dagster_dbt.cli.resources.DbtCliResource`` ? If I don’t wrap it up with

@resource

, It is of correct type and works. But I was hoping to instantiate the resource dynamically

Hi madhurt! So in your

dbt_resource

function, instead of returning a configured

dbt_cli_resource

(which is a resource definition), I think you want to return an instance of

DbtCliResource

, which is the wrapper around the dbt cli. You can pass in the configuration directly: https://docs.dagster.io/_apidocs/libraries/dagster-dbt#dagster_dbt.DbtCliResource

another solution here is to write your own op that uses the resource. that would basically look like the example here: https://docs.dagster.io/integrations/dbt#-supplying-an-argument, but instead of setting the

models

argument, you could set the

project_dir

argument

@prha How can one instantiate an object of

DbtCliResource

? I am thinking of

return DbtCliResource(
        executable=context.resource_config["dbt_executable"],
        default_flags={"project_dir": dbt_project_dir, "profiles_dir": dbt_project_dir},
        warn_error=context.resource_config["warn_error"],
        ignore_handled_error=context.resource_config["ignore_handled_error"],
        target_path=context.resource_config["target_path"],
        logger=context.log,
    )