Metadata-Version: 2.4
Name: flyteplugins-dbt
Version: 2.11.4
Summary: dbt task plugin for flyte
Author-email: Jan Fiedler <jan@union.ai>
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: dbt-core
Requires-Dist: flyte

# 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.

```python
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:

```python
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`.

```python
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:

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