Metadata-Version: 2.4
Name: fast2dev-dbt-runner
Version: 0.0.1
Summary: Airflow 3 operators that turn a dbt 1.12 manifest into DAGs for Fast2.Dev (k8s, bash, API, and GKE).
Author-email: M7XLab <support@fast2.dev>
Maintainer-email: "Fast2.Dev" <support@fast2.dev>
License: MIT
Project-URL: Homepage, https://fast2.dev
Project-URL: Documentation, https://fast2.dev
Project-URL: Repository, https://gitea.m7xlab.top/fast2dev/fast2dev-ddp
Project-URL: Bug Tracker, https://gitea.m7xlab.top/fast2dev/fast2dev-ddp/issues
Project-URL: Changelog, https://gitea.m7xlab.top/fast2dev/fast2dev-ddp/src/branch/main/services/dbt/workflow-runner/CHANGELOG.md
Keywords: airflow,airflow3,dbt,dbt-core,fast2dev,m7xlab,data-orchestration,kubernetes,gke,manifest-parser,workflow,etl,data-engineering
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Scientific/Engineering :: Information Analysis
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Framework :: Apache Airflow
Requires-Python: <3.13,>=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: kubernetes>=18.0.0
Requires-Dist: google-cloud-storage>=2.0.0
Requires-Dist: google-auth>=2.26.1
Requires-Dist: requests>=2.25.0
Requires-Dist: pyyaml>=5.4.0
Requires-Dist: jinja2>=3.0.0
Provides-Extra: airflow
Requires-Dist: apache-airflow<4,>=3.0.0; extra == "airflow"
Requires-Dist: apache-airflow-providers-cncf-kubernetes>=10.0.0; extra == "airflow"
Provides-Extra: dev
Requires-Dist: black>=21.0.0; extra == "dev"
Requires-Dist: build>=1.0.0; extra == "dev"
Requires-Dist: flake8>=3.8.0; extra == "dev"
Requires-Dist: mypy>=0.800; extra == "dev"
Requires-Dist: pre-commit>=2.15.0; extra == "dev"
Requires-Dist: pytest>=6.0.0; extra == "dev"
Requires-Dist: pytest-cov>=2.10.0; extra == "dev"
Requires-Dist: twine>=5.0.0; extra == "dev"
Requires-Dist: wheel>=0.37.0; extra == "dev"
Provides-Extra: docs
Requires-Dist: sphinx>=4.0.0; extra == "docs"
Requires-Dist: sphinx-rtd-theme>=1.0.0; extra == "docs"
Requires-Dist: myst-parser>=0.15.0; extra == "docs"
Dynamic: license-file

# fast2dev-dbt-runner

[![PyPI version](https://img.shields.io/pypi/v/fast2dev-dbt-runner.svg)](https://pypi.org/project/fast2dev-dbt-runner/)
[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Airflow 3](https://img.shields.io/badge/airflow-3.x-brightgreen.svg)](https://airflow.apache.org/)
[![dbt 1.12](https://img.shields.io/badge/dbt-1.12-orange.svg)](https://docs.getdbt.com/)

Airflow 3 library that reads a dbt 1.12 `manifest.json` and builds DAGs for the [Fast2.Dev](https://fast2.dev) data platform. M7XLab publishes the package; worker images come from the private registry `registry.m7xlab.top/fast2dev/*`, not Docker Hub.

This is a **library**. There are no `console_scripts` entry points. Install it into the Airflow environment and import `fast2dev_dbt_runner`.

## What it does

Four execution backends, same manifest parser:

| Operator | Import alias | Best for | Worker image |
|----------|--------------|----------|--------------|
| `k8s` | `DbtManifestParserK8sOperator` | Daily jobs, high concurrency, lowest cost | `registry.m7xlab.top/fast2dev/dbt-workflow-core` |
| `bash` | `DbtManifestParserBashOperator` | Medium projects on Airflow workers | Airflow worker image |
| `api` | `DbtManifestParserApiOperator` | Low-latency runs against a pre-warmed dbt API | Fast2.Dev dbt api-server |
| `gke` | `DbtManifestParserGkeOperator` | Isolated external-client clusters | `registry.m7xlab.top/fast2dev/dbt-workflow-core` |

DAG templates in the Fast2.Dev platform import the parser classes from the operator modules directly (for example `from fast2dev_dbt_runner.dbt_manifest_parser_k8s_operator import DbtManifestParser`). The aliases above are the public package-level names.

## Requirements

- Python 3.10–3.12
- Apache Airflow 3.x (install the `[airflow]` extra, or rely on the platform image)
- dbt Core 1.12 manifests (schema v12)
- Private registry access for k8s/GKE images (`registry.m7xlab.top`)

## Installation

```bash
pip install fast2dev-dbt-runner

# Airflow 3 + Kubernetes provider
pip install fast2dev-dbt-runner[airflow]

# Local development (tests, linters, twine)
pip install fast2dev-dbt-runner[dev]
```

From the Fast2.Dev monorepo (this package is not a standalone GitHub repository):

```bash
git clone git@gitea.m7xlab.top:fast2dev/fast2dev-ddp.git
cd fast2dev-ddp/services/dbt/workflow-runner
pip install -e ".[dev]"
```

Publishing to pypi.org is documented in [docs/publishing.md](docs/publishing.md). Do not commit a PyPI token or a `.pypirc` file.

## Usage

```python
from airflow.sdk import DAG
from fast2dev_dbt_runner import DbtManifestParserK8sOperator

with DAG(
    dag_id="analytics_dbt",
    schedule="@daily",
    catchup=False,
) as dag:
    parser = DbtManifestParserK8sOperator(...)
```

Platform-generated DAGs live in `services/cicd/runner` and already import `fast2dev_dbt_runner`. Configure behavior with Airflow Variables, not by editing this library.

## Airflow Variable Reference

All variables are read from Airflow Variables at DAG load time. Defaults shown in parentheses.

### Infrastructure & Identity

| Variable | Default | Description |
|----------|---------|-------------|
| `PROJECT_ID` | required | Cloud project identifier |
| `DBT_PROJECT_NAME` | required | dbt project name (used as DAG ID prefix) |
| `NAMESPACE` | — | Kubernetes namespace (k8s/GKE operators) |
| `DAG_OWNER` | `fast2.dev` | Airflow DAG owner |
| `DAG_START_DATE` | `days_ago(1)` | DAG start date expression |
| `DAG_SCHEDULE_INTERVAL` | `@once` | Cron expression or preset (`@daily`, `@hourly`, etc.) |
| `GIT_BRANCH` | — | Git branch to checkout on the worker before running dbt |
| `IMAGE` | platform default | dbt worker image, typically `registry.m7xlab.top/fast2dev/dbt-workflow-core:<tag>` |

### Model Execution Control

These three flags follow the same pattern: `is_in_manifest` → `DBT_X` → `DBT_X_SHARDING`.

| Variable | Default | Description |
|----------|---------|-------------|
| `DBT_MODEL` | `True` | Enable/disable model (`dbt run`) execution entirely |
| `DBT_MODEL_SHARDING` | `True` | `True` = one Airflow task per model with full dependency lineage; `False` = single batch task running `dbt run --select "model1 model2 ..."` |

### Seed Execution Control

| Variable | Default | Description |
|----------|---------|-------------|
| `DBT_SEED` | `False` | Enable/disable seed (`dbt seed`) execution |
| `DBT_SEED_SHARDING` | `True` | `True` = one task per seed file; `False` = single batch task with `--select` |

### Snapshot Execution Control

| Variable | Default | Description |
|----------|---------|-------------|
| `DBT_SNAPSHOT` | `False` | Enable/disable snapshot (`dbt snapshot`) execution |
| `DBT_SNAPSHOT_SHARDING` | `True` | `True` = one task per snapshot; `False` = single batch task with `--select` |
| `DBT_MODEL_DEPENDS_ON_SNAPSHOT` | `False` | When `True`, a `model -> snapshot -> model` chain (a snapshot a model depends on via `ref()`) runs in correct order within a single DAG. **Sharded** (`DBT_MODEL_SHARDING=True`): the snapshot (and its tests) is woven into the model task group as task-level dependencies. **Batch** (`DBT_MODEL_SHARDING=False`): the model-run + snapshot batches are replaced by a single `dbt build --select "<models> <snapshots>"`, which orders models, snapshots, and tests by lineage in one process. No-op when no model actually depends on a snapshot. Supported across all operators (bash, k8s, GKE, API). |

### Source Freshness Control

| Variable | Default | Description |
|----------|---------|-------------|
| `DBT_SOURCE` | `True` | Enable/disable source freshness (`dbt source freshness`) checks |
| `DBT_SOURCE_SHARDING` | `True` | `True` = one task per source; `False` = single batch task with `--select` |

### Pipeline Steps

| Variable | Default | Description |
|----------|---------|-------------|
| `DBT_DEPS` | `True` | Run `dbt deps` at DAG start to install packages. Set `False` when packages are vendored in the repo |
| `DATA_QUALITY` | `False` | Enable re_data / data quality task at end of DAG |
| `DEBUG` | `False` | Run `dbt debug` at DAG start to verify connection |

### Filtering & Selection

| Variable | Default | Description |
|----------|---------|-------------|
| `DBT_TAGS` | — | Comma-separated list of dbt tags. Only nodes tagged with **all** listed tags are included. Example: `tag1,tag2` |

### Full Refresh

| Variable | Default | Description |
|----------|---------|-------------|
| `FULL_REFRESH` | `False` | Run models with `dbt run --full-refresh` (rebuilds incrementals from scratch) |
| `FULL_REFRESH_MODEL_NAME` | — | Comma-separated list of specific model names to full-refresh (others run normally) |

### E2E / Testing Modes

| Variable | Default | Description |
|----------|---------|-------------|
| `E2E_MODE_EMPTY` | `False` | Append `--empty` to `dbt run` (creates empty tables without processing data, for E2E schema validation) |

### Monitoring & Logging

| Variable | Default | Description |
|----------|---------|-------------|
| `MODEL_DEBUG_LOG` | `False` | Log compiled SQL for failed model tasks (appends compiled code to Airflow task logs) |
| `DATAHUB_ENABLED` | `False` | Enable DataHub metadata push after DAG run |

### Manifest Cache (Environment Variables)

These are set as environment variables on the Airflow workers, not as Airflow Variables:

| Variable | Default | Description |
|----------|---------|-------------|
| `AIRFLOW__CORE__MANIFEST_CACHE_ENABLED` | `True` | Enable manifest caching (reduces DAG parse time by 99%+ for unchanged manifests) |
| `AIRFLOW__CORE__MANIFEST_CACHE_DEBUG` | `False` | Log cache hit/miss details |
| `AIRFLOW__CORE__MANIFEST_CACHE_MAX_SIZE` | `50` | Maximum number of manifests to keep in the module-level cache |

---

## Sharding Explained

### Sharding = True (default) — Full Lineage

Each model/seed/snapshot/source becomes its own Airflow task. Airflow builds the full dependency graph from the dbt manifest, so tasks execute in dependency order and you can retry individual failed nodes.

```
seed_customers → model_stg_customers → model_orders → model_revenue
                                     ↗
                model_stg_orders ───
```

**Use when**: You need visibility into individual model failures, want to retry a single model, or have long-running models that benefit from parallelism.

### Sharding = False — Batch Mode

All tag-filtered models are collected from the manifest and passed in a single `dbt run --select "model1 model2 ..."` command, running as one Airflow task. Tag filtering (`DBT_TAGS`) is fully respected — only manifested, tag-matching nodes are included.

```
single_task: dbt run --select "stg_customers stg_orders dim_revenue"
```

**Use when**: You have many small models, want simpler DAGs with fewer tasks, or Airflow overhead per-task is significant.

---

## Configuration Examples

### Standard Daily Pipeline (K8S)

```python
# Airflow Variables
{
    "PROJECT_ID": "my-gcp-project",
    "DBT_PROJECT_NAME": "analytics",
    "DAG_SCHEDULE_INTERVAL": "@daily",
    "DBT_DEPS": "True",
    "DBT_SOURCE": "True",
    "DBT_SOURCE_SHARDING": "True",
    "DBT_SEED": "False",
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "True",
    "DBT_SNAPSHOT": "False",
    "DATA_QUALITY": "True"
}
```

### Batch Mode (Many Small Models, Low Overhead)

```python
# All resource types run as single batch tasks — fewer Airflow tasks, simpler DAG
{
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "False",   # dbt run --select "model1 model2 ..."
    "DBT_SEED": "True",
    "DBT_SEED_SHARDING": "False",    # dbt seed --select "seed1 seed2 ..."
    "DBT_SOURCE": "True",
    "DBT_SOURCE_SHARDING": "False",  # dbt source freshness --select "src1 src2 ..."
    "DBT_SNAPSHOT": "True",
    "DBT_SNAPSHOT_SHARDING": "False" # dbt snapshot --select "snap1 snap2 ..."
}
```

### Tag-Filtered Pipeline

```python
# Only run nodes tagged with both "marketing" and "daily"
{
    "DBT_TAGS": "marketing,daily",
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "True"
}
```

### Full Refresh Specific Models

```python
# Full refresh only two models; others run normally
{
    "FULL_REFRESH": "False",
    "FULL_REFRESH_MODEL_NAME": "dim_customers,fct_orders"
}
```

### Full Refresh All Models

```python
{
    "FULL_REFRESH": "True"
}
```

### E2E Schema Validation

```python
# Creates empty tables (no data) to validate schema changes end-to-end
{
    "E2E_MODE_EMPTY": "True",
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "False"
}
```

### Skip Package Installation (Vendored Packages)

```python
# Packages are committed to the repo — skip dbt deps for faster, more reliable runs
{
    "DBT_DEPS": "False"
}
```

### High-Performance Real-Time Pipeline (API Operator)

```python
{
    "PROJECT_ID": "my-gcp-project",
    "DBT_PROJECT_NAME": "realtime_analytics",
    "DAG_SCHEDULE_INTERVAL": "*/15 * * * *",
    "DBT_DEPS": "False",
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "False",  # batch mode for speed
    "DBT_TAGS": "realtime",
    "MODEL_DEBUG_LOG": "True"
}
```

---

## Architecture

### Execution Flow per DAG

```
[Airbyte sync] (optional)
     ↓
[dbt deps]     (if DBT_DEPS=True)
     ↓
[dbt debug]    (if DEBUG=True)
     ↓
[show_input_data]
     ↓
[dbt source freshness]  (if DBT_SOURCE=True)
     ↓
[dbt seed]              (if DBT_SEED=True)
     ↓
[dbt run]               (if DBT_MODEL=True)
     ↓
[dbt snapshot]          (if DBT_SNAPSHOT=True)
     ↓
[re_data / quality]     (if DATA_QUALITY=True)
```

### Manifest Caching

The manifest caching system reduces DAG import time by 99%+ for unchanged manifests:

- **Before caching**: ~2–4 seconds per manifest parse, ~480 parses/hour with 2 schedulers
- **After caching**: <10ms for cache hits, only 5–10 cache misses/hour (on actual manifest changes)
- Cache keys include: file MD5 hash + `DBT_TAGS` + ancestor/descendant flags
- Cache is process-local (not shared across pod restarts); first parse after restart is always a cache miss

---

## Development

```bash
pytest tests/
pytest tests/ --cov=fast2dev_dbt_runner --cov-report=term-missing
```

See [CONTRIBUTING.md](CONTRIBUTING.md) and [docs/publishing.md](docs/publishing.md) for release steps. Version **0.0.1** is the first public PyPI name under M7XLab / Fast2.Dev. Prior calendar versions (for example `2026.1.0.9`) were internal.

## Support

- **Homepage**: [fast2.dev](https://fast2.dev)
- **Source**: [gitea.m7xlab.top/fast2dev/fast2dev-ddp](https://gitea.m7xlab.top/fast2dev/fast2dev-ddp) (`services/dbt/workflow-runner`)
- **Email**: support@fast2.dev
- **Changelog**: [CHANGELOG.md](CHANGELOG.md)
