Metadata-Version: 2.4
Name: acceldata-openlineage
Version: 1.0.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. Install this package in your
Airflow environment and register the extractors so lineage events include
Acceldata-specific facets for dbt Cloud runs.

## Prerequisites

OpenLineage should already be enabled and configured in your Airflow
environment. This package additionally requires:

| Component | Notes |
|-----------|--------|
| Apache Airflow | 2.10 or later |
| `apache-airflow-providers-openlineage` | Required for custom extractors |
| `apache-airflow-providers-dbt-cloud` | Required for `DbtCloudRunJobOperator` |

This package adds one runtime dependency: `attrs`.

## Install

Install on every Airflow component that runs or schedules tasks (workers,
scheduler, and triggerer if applicable):

```bash
pip install acceldata-openlineage
```

**Self-managed Airflow:** install into the same Python environment used by
Airflow processes, then restart workers and the scheduler.

For **Google Cloud Composer**, follow the dedicated section below.

## Setup on Google Cloud Composer

These steps assume OpenLineage is already configured in your Composer
environment. Changes are made in the [Google Cloud Console](https://console.cloud.google.com/)
under **Composer → Environments → your environment → Edit**.

### 1. Add PyPI packages

Open the **PyPI packages** tab and add:

| Package | Notes |
|---------|--------|
| `acceldata-openlineage` | This package (pin a version in production, e.g. `==1.0.0`) |
| `apache-airflow-providers-dbt-cloud` | Required if not already installed |

Composer installs packages on all Airflow components during the next environment
update.

### 2. Register the extractor

Open the **Environment variables** tab and set `OPENLINEAGE_EXTRACTORS`
(`AIRFLOW__OPENLINEAGE__EXTRACTORS` is an equivalent alias).

Append this extractor to any extractors you already use—do not replace the
full list unless you intend to:

```
acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor
```

Example when other extractors are already registered:

```
your.existing.Extractor;acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor
```

### 3. Configure the dbt Cloud connection

In the Airflow UI (**Admin → Connections**), create or update the connection
used by your DAGs:

| Field | Value |
|-------|--------|
| Connection Id | Same as `dbt_cloud_conn_id` on the operator (default: `dbt_cloud_default`) |
| Connection Type | `dbt Cloud` |
| Login | dbt Cloud **Account ID** |
| Password | dbt Cloud API token |

When `account_id` is not set on the operator, the extractor resolves
`dbtCloudAccountId` from the connection login.

### 4. Apply the environment update

Save the Composer environment changes. Composer rebuilds workers and the
scheduler; this can take several minutes.

After the update completes, run a DAG with a `DbtCloudRunJobOperator` task and
check task logs for `Attached dbtCloud facet`.

## Register extractors (other environments)

For self-managed Airflow, MWAA, Astronomer, or other platforms, register the
extractor via environment variable or `airflow.cfg`.

Set a semicolon-separated list of fully qualified class names. Append this
extractor to any extractors you already use:

**Environment variable:**

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

If you already have extractors configured, append with `;`:

```bash
export AIRFLOW__OPENLINEAGE__EXTRACTORS=\
your.existing.Extractor;\
acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor
```

**`airflow.cfg`:**

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

`OPENLINEAGE_EXTRACTORS` is an equivalent alias supported by the OpenLineage
provider.

Restart Airflow workers and the scheduler after changing extractor configuration.

### dbt Cloud connection

Configure a dbt Cloud Airflow connection as usual:

| Connection field | Value |
|------------------|--------|
| Conn Id | Same as `dbt_cloud_conn_id` on the operator (default: `dbt_cloud_default`) |
| Login | dbt Cloud **Account ID** |
| Password | dbt Cloud API token |

When `account_id` is not set on the operator, the extractor resolves
`dbtCloudAccountId` from the connection login (same behavior as the dbt Cloud
provider hook).

## Included extractors

| Extractor | Operator | Facet key |
|-----------|----------|-----------|
| `DbtCloudRunJobOperatorExtractor` | `DbtCloudRunJobOperator` | `dbtCloud` |

### `dbtCloud` run facet

Attached when `operator.run_id` is set (normally on COMPLETE / FAIL; also on
START if the run was reused). Fields:

| Field | Source |
|-------|--------|
| `dbtCloudRunId` | `DbtCloudRunJobOperator.run_id` |
| `dbtCloudJobId` | `DbtCloudRunJobOperator.job_id` |
| `dbtCloudAccountId` | `DbtCloudRunJobOperator.account_id`, or the Account ID (`login`) from `dbt_cloud_conn_id` when unset |

The extractor delegates to the operator's built-in OpenLineage methods so
default dbt Cloud lineage behavior is preserved, then merges the Acceldata
facet used for InterPipelineRunLink.

## Verify installation

On an Airflow worker node, confirm the package and extractor import correctly:

```python
from acceldata_openlineage import DbtCloudRunJobOperatorExtractor

print(DbtCloudRunJobOperatorExtractor.get_operator_classnames())
# ['DbtCloudRunJobOperator']
```

Run a DAG with a `DbtCloudRunJobOperator` task and confirm lineage events
include a `dbtCloud` run facet with `dbtCloudRunId`, `dbtCloudJobId`, and
`dbtCloudAccountId` populated.

## Troubleshooting

| Symptom | What to check |
|---------|----------------|
| Extractor not loading | Import path matches config exactly; package installed on the worker running the task |
| No `dbtCloud` facet on START | Expected when `run_id` is not set yet; facet appears on COMPLETE / FAIL |
| Missing `dbtCloudAccountId` | Set `account_id` on the operator, or put the Account ID in the connection **Login** field |
| Composer: changes not picked up | Environment update finished successfully |

Check Airflow task logs for messages such as `Attached dbtCloud facet` or
warnings about connection lookup failures.
