Hello everyone! I'm trying to get started using

dagster_dbt

. I'm encountering the following error:

agster._core.errors.DagsterInvalidDefinitionError: resource with key 'dbt' required by op 'run_dbt_da47d' was not provided. Please provide a <class 'dagster._core.definitions.resource_definition.ResourceDefinition'> to key 'dbt', or change the required key to one of the following keys which points to an <class 'dagster._core.definitions.resource_definition.ResourceDefinition'>: ['io_manager']

I see that

dbt_cli_resource.configured(
        {"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR}
    )

Is already a ResourceDefinition.

import os

from dagster_dbt import dbt_cli_resource, load_assets_from_dbt_project
from dagster_duckdb_pandas import duckdb_pandas_io_manager

from dagster import (
    Definitions,
    ScheduleDefinition,
    define_asset_job,
    fs_io_manager,
    load_assets_from_package_module,
)
from dagster._utils import file_relative_path

DBT_PROJECT_DIR = file_relative_path(__file__, "../dbt_project")
DBT_PROFILES_DIR = file_relative_path(__file__, "../dbt_project/config")

# all assets live in the default dbt_schema
dbt_assets = load_assets_from_dbt_project(
    DBT_PROJECT_DIR,
    DBT_PROFILES_DIR,
    # prefix the output assets based on the database they live in plus the name of the schema
    key_prefix=["duckdb", "dbt_schema"],
    # prefix the source assets based on just the database
    # (dagster populates the source schema information automatically)
    source_key_prefix=["duckdb"],
)

a = dbt_cli_resource.configured(
        {"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR})

print(type(a))

resources = {
    # this io_manager allows us to load dbt models as pandas dataframes
    "io_manager": duckdb_pandas_io_manager.configured(
        {"database": os.path.join(DBT_PROJECT_DIR, "example.duckdb")}
    ),
    # this resource is used to execute dbt cli commands
    "dbt": dbt_cli_resource.configured(
        {"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR}
    )
}

everything_job = define_asset_job("everything_everywhere_job", selection="*")

defs = Definitions(
    assets=[*dbt_assets],
    resources=resources,
    schedules=[
        ScheduleDefinition(job=everything_job, cron_schedule="@weekly")    ]
)

Can someone please point out what am I missing?

Hey @Constantin Lungu how are you materializing the dbt assets (through the ui, cli, etc)? also cc @rex @owen for visibility

Hi, I’m unable to replicate this. I edited the

tutorial_dbt_dagster

project with your sample code and I’m not getting your error.

dagster project from-example --name tutorial_dbt_dagster --example tutorial_dbt_dagster

import os
from dagster import (
    Definitions,
    load_assets_from_modules,
    define_asset_job,
    ScheduleDefinition,
)
from dagster_dbt import dbt_cli_resource
from dagster_duckdb_pandas import duckdb_pandas_io_manager

from tutorial_dbt_dagster import assets
from tutorial_dbt_dagster.assets import DBT_PROFILES, DBT_PROJECT_PATH

resources = {
    "dbt": dbt_cli_resource.configured(
        {
            "project_dir": DBT_PROJECT_PATH,
            "profiles_dir": DBT_PROFILES,
        },
    ),
    "io_manager": duckdb_pandas_io_manager.configured(
        {"database": os.path.join(DBT_PROJECT_PATH, "tutorial.duckdb")}
    ),
}

everything_job = define_asset_job("everything_everywhere_job", selection="*")

defs = Definitions(
    assets=load_assets_from_modules([assets]),
    resources=resources,
    schedules=[ScheduleDefinition(job=everything_job, cron_schedule="@weekly")],
)

also what version of dagster are you using? there was a bug a while back that made use erroneously determine which resources were available in some very specific circumstances

This is the version of my packages. So

dagster-1.2.2

ok - the bug would have been fixed by then. I think it was back in like 1.1.7 or something

@rex Yes I did reload code definitions, result still the same.