flyteplugins-dbt

Run dbt CLI invocations as Flyte v2 tasks.

DbtTask maps one dbtRunner.invoke(...) call to one Flyte task. Project-level dbt paths are configured on the task, and invocation-level options are task inputs.

Install the dbt adapter required by your project, such as dbt-duckdb, dbt-bigquery, or dbt-snowflake, in the task image alongside this plugin.

from pathlib import Path

import flyte
from flyteplugins.dbt import DbtTask

DBT_PROJECT_DIR = "jaffle_shop"
DBT_PROFILES_DIR = "dbt-profiles"

env = flyte.TaskEnvironment(
    name="dbt",
    image=flyte.Image.from_debian_base()
    .with_requirements("requirements.txt")
    .with_source_folder(Path(DBT_PROJECT_DIR))
    .with_source_folder(Path(DBT_PROFILES_DIR)),
)

dbt_test = DbtTask(
    name="dbt-test",
    task_environment=env,
    project_dir=DBT_PROJECT_DIR,
    profiles_dir=DBT_PROFILES_DIR,
    profile=DBT_PROJECT_DIR,
    report=True,
)


@env.task
async def main():
    return await dbt_test.aio(command="test", select=["stg_orders+"], target="prod")

Multi-token dbt commands can be passed as a shell-style string or as explicit tokens:

await dbt_test.aio(command="docs generate")
await dbt_test.aio(command=["source", "freshness"])

The task returns a list of DbtNodeResult values summarized from dbt node results. If dbt reports failure, the task raises the dbt exception when one is available, otherwise it raises DbtInvocationError with the summarized node results.

When report=True, the task writes a dbt report tab with node names, resource types, statuses, failures, execution times, and messages. The report is written before dbt failures are raised, so failed dbt runs can still show succeeded and failed node results.

Event callbacks

Custom dbt event callbacks can be passed to DbtTask.

def log_dbt_node(event):
    data = getattr(event, "data", None)
    node_info = getattr(data, "node_info", None)
    if node_info is None:
        return

    node_name = getattr(node_info, "node_name", None)
    node_status = getattr(node_info, "node_status", None)
    print(f"dbt node={node_name} status={node_status}")


dbt_test = DbtTask(
    name="dbt-test",
    task_environment=env,
    callbacks=[log_dbt_node],
)

For remote execution, callbacks must be importable functions. Import-path strings are also supported:

dbt_test = DbtTask(
    name="dbt-test",
    task_environment=env,
    callbacks=["my_project.callbacks.log_dbt_node"],
)