Hello! I'm trying to integrate Dagster with dbt and I am getting the following error when trying to run my test job.

dagster._check.CheckError: DuckDBIOManager does not have a handler for type '<class 'dagster_dbt.core.types.DbtCliOutput'>'. Has handlers for types '<class 'polars.dataframe.frame.DataFrame'>'. Please build the DuckDBIOManager with an type handler for type '<class 'dagster_dbt.core.types.DbtCliOutput'>', so the DuckDBIOManager can correctly handle the output.

Code:

import os

from dagster_dbt import dbt_cli_resource, dbt_test_op, load_assets_from_dbt_project

from dagster_duckdb_polars import duckdb_polars_io_manager

from dagster import Definitions, file_relative_path
from dagster import job

DBT_PROJECT_PATH = file_relative_path(__file__, "../dbt")
DBT_PROFILES = file_relative_path(__file__, "../dbt/config")


dbt_resource = dbt_cli_resource.configured(
        {
            "project_dir": DBT_PROJECT_PATH,
            "profiles_dir": DBT_PROFILES,
        }
    )

model_resources = {
    "dbt": dbt_resource,
    "io_manager": duckdb_polars_io_manager.configured(
        {"database": "/shared/db/datamart.duckdb"})
}



@job(resource_defs={'dbt': dbt_resource})
def run_dbt_test_job():
    dbt_test_op()

defs = Definitions(assets=load_assets_from_dbt_project(DBT_PROJECT_PATH, profiles_dir=DBT_PROFILES, use_build_command=True), resources=model_resources, jobs=[run_dbt_test_job])

Any idea where the problem might lie? Thank you!

@jamie could you take a look at this io manager issue?

this would happen if the dbt_assets output was getting handled by the IO Manager - i thought dbt assets bypassed the IO manager though. looping @owen in because we had a similar report the other day

Looks like the error is happening from an interaction between the io manager and the default output emitted by

dbt_test_op

(

DbtCliOutput

) You have two options: 1. Remove your usage of

duckdb_polars_io_manager

. 2. Migrate your usage of

dbt_cli_resource

to

DbtCliResource

, and invoke the op manually. Here's an example: https://dagster.slack.com/archives/C04CW71AGBW/p1693316175875669?thread_ts=1693311924.216999&cid=C04CW71AGBW You'll want a code snippet similar to the following

@op
def dbt_test_op(dbt: DbtCliResource):
    dbt.cli(["test"], manifest={}).wait()

@job
def run_dbt_test_job():
    dbt_test_op()

Thanks @rex, that solved it for me!