Metadata-Version: 2.4
Name: acceldata-openlineage
Version: 1.1.0
Summary: Custom OpenLineage extractors for Acceldata integrations
Author: Acceldata
License: Apache-2.0
Keywords: openlineage,airflow,lineage,extractor,dbt,acceldata
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: License :: OSI Approved :: Apache Software License
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: attrs>=23
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: pytest-mock; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Requires-Dist: black; extra == "dev"
Provides-Extra: release
Requires-Dist: build>=1; extra == "release"
Requires-Dist: twine>=5; extra == "release"

# Acceldata OpenLineage

Custom OpenLineage extractors for Acceldata. Adds a **`dbtCloud` run facet** to
`DbtCloudRunJobOperator` lineage events so ADOC can link Airflow tasks to dbt Cloud
runs (InterPipelineRunLink).

By default, the extractor also delegates to Airflow's built-in dbt Cloud OpenLineage
integration. If the **Acceldata dbt Cloud Data Source connector** already collects
detailed dbt lineage, you can disable that delegation (see below).

## Prerequisites

OpenLineage must already be enabled in Airflow, plus:

| Component | Notes |
|-----------|--------|
| Apache Airflow | 2.10+ |
| `apache-airflow-providers-openlineage` | Custom extractors |
| `apache-airflow-providers-dbt-cloud` | `DbtCloudRunJobOperator` |

## Install

Install on workers, scheduler, and triggerer (if used):

```bash
pip install acceldata-openlineage==1.1.0
```

Restart Airflow processes after install.

## Setup

### 1. Register the extractor

Append to existing extractors (semicolon-separated); do not replace the full list
unless intended:

```ini
# airflow.cfg
[openlineage]
extractors = acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor
```

Or via environment variable:

```bash
AIRFLOW__OPENLINEAGE__EXTRACTORS=acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor
```

**Composer:** add the package under **PyPI packages**, set
`AIRFLOW__OPENLINEAGE__EXTRACTORS` under **Environment variables**, then apply
the environment update.

### 2. dbt Cloud connection

| Field | Value |
|-------|--------|
| Conn Id | Same as `dbt_cloud_conn_id` (default: `dbt_cloud_default`) |
| Login | dbt Cloud **Account ID** |
| Password | dbt Cloud API token |

If `account_id` is unset on the operator, `dbtCloudAccountId` is taken from the
connection login.

### 3. Verify

Run a `DbtCloudRunJobOperator` task and check logs for `Attached dbtCloud facet`.
On COMPLETE / FAIL the facet should include `dbtCloudRunId`, `dbtCloudJobId`, and
`dbtCloudAccountId`.

## Built-in dbt artifact extraction (optional)

**1.1.0+.** The setting `dbt_cloud_lineage_extraction_enabled` controls whether
the extractor delegates to Airflow's built-in dbt Cloud OpenLineage methods
(`get_openlineage_facets_on_*`).

| Value | Behavior |
|-------|----------|
| `true` (**default**) | Built-in extraction runs (same as prior releases). |
| `false` | Built-in extraction is skipped. Task-level events and the `dbtCloud` facet are still emitted. |

Set to `false` when the Acceldata dbt Cloud Data Source connector already
collects model/test/dataset lineage in ADOC and you only need orchestration
lineage plus the `dbtCloud` linking facet from Airflow.

```ini
[openlineage]
dbt_cloud_lineage_extraction_enabled = false
```

```bash
AIRFLOW__OPENLINEAGE__DBT_CLOUD_LINEAGE_EXTRACTION_ENABLED=false
```

Restart scheduler and workers after changing this.

Note: built-in extraction also requires `wait_for_termination=True` on the
operator. With `wait_for_termination=False`, artifact extraction is already a
no-op regardless of this flag.

## `dbtCloud` facet

| Field | Source |
|-------|--------|
| `dbtCloudRunId` | `operator.run_id` |
| `dbtCloudJobId` | `operator.job_id` |
| `dbtCloudAccountId` | `operator.account_id`, or connection login |

Present on COMPLETE / FAIL once `run_id` is set (START only if the run was reused).

## Troubleshooting

- **No facet** — confirm the extractor path in config, package on the worker, and
  task reached COMPLETE / FAIL with a `run_id`.
- **Missing `dbtCloudAccountId`** — set `account_id` on the operator or Account ID
  in the connection login.
- **Flag has no effect** — set `wait_for_termination=True` to test; with `False`,
  built-in extraction never runs anyway.
- **Duplicate lineage or upstream errors** (`KeyError: 'host'`, `catalog.json`
  404) — set `dbt_cloud_lineage_extraction_enabled = false`.
