Metadata-Version: 2.4
Name: dpone-airflow-pack
Version: 0.66.12
Summary: Lightweight Airflow scheduler-side pack provider for dpone GitOps workloads
Keywords: airflow,gitops,data-engineering,scheduler,kubernetes
Author: PaulKov
License-Expression: Apache-2.0
Classifier: Development Status :: 3 - Alpha
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Dist: pyyaml>=6.0.1
Maintainer: PaulKov
Requires-Python: >=3.11, <3.13
Project-URL: Homepage, https://github.com/PaulKov/dpone
Project-URL: Repository, https://github.com/PaulKov/dpone
Project-URL: Issues, https://github.com/PaulKov/dpone/issues
Project-URL: Documentation, https://paulkov.github.io/dpone/
Description-Content-Type: text/markdown

# dpone-airflow-pack

`dpone-airflow-pack` is the lightweight Airflow scheduler/webserver provider for dpone GitOps packs.

It only reads a static `airflow-pack.json` and builds visible Airflow/Kubernetes tasks. It does not import the full
`dpone` runtime and intentionally contains no source/sink/native transfer dependencies such as ClickHouse, MSSQL,
`pyodbc`, pandas, polars, or ConnectorX.

Recommended Airflow DAG import:

```python
from dpone_airflow_pack import build_dpone_gitops_task_group_from_pack
```

The full `dpone[full,accel]` package belongs in the KPO runtime image, not in the scheduler image.

## Remote pack cache

Remote `airflow-pack.json` artifacts can be published to object storage and
mirrored into a bounded local scheduler cache.

Bootstrap one generation:

```bash
dpone-airflow-pack-sync \
  --once \
  --index-uri s3://do-dwh/dpone-artifacts/prod/airflow-dags-dev/latest/pack-index.json \
  --reader-connection-id s3_dpone_artifacts_reader \
  --cache-dir /opt/airflow/dags/.dpone-cache/airflow \
  --status-path /opt/airflow/dags/.dpone-cache/airflow/status/last-sync-status.json \
  --airflow-variable-key dpone_airflow_pack_cache_status
```

Refresh without restarting Airflow pods:

```bash
dpone-airflow-pack-sync \
  --watch \
  --interval-seconds 60 \
  --jitter-seconds 15 \
  --index-uri s3://do-dwh/dpone-artifacts/prod/airflow-dags-dev/latest/pack-index.json \
  --reader-connection-id s3_dpone_artifacts_reader \
  --cache-dir /opt/airflow/dags/.dpone-cache/airflow \
  --status-path /opt/airflow/dags/.dpone-cache/airflow/status/last-sync-status.json \
  --airflow-variable-key dpone_airflow_pack_cache_status
```

The sync loop is fail-open. On S3, connection, index, or hash errors it writes a
warning status and preserves the previous `current` generation. DAG parse stays
local and cache-only.

When `--airflow-variable-key` is set, the sync process also publishes the same
redacted status payload into Airflow metadata. This is the recommended
diagnostic channel for KubernetesExecutor deployments because runtime worker
pods do not need to mount the scheduler or dagProcessor cache volume. Variable
publish tries the Airflow model first and then a direct metadata DB upsert for
sidecar/init contexts where Airflow 3 task SDK variables are not available.
Publish failures are warnings only and never fail the Airflow pod.

Connection resolution is scheduler-side safe for Airflow 2.x and 3.x:
environment connection URI first, Airflow hook second, and metadata database
fallback third. The metadata fallback matters for Airflow 3 sidecar/init
containers, where provider hooks may not resolve DB-backed connections outside
task execution context even though the metadata row exists.

## GitOps domain groups

For DAGs that should run a whole GitOps domain or a named workflow group, keep
the grouping in the domain catalog and resolve it through the lightweight
provider:

```yaml
workflow_groups:
  daily:
    workload_ids:
      - dim_customer
      - fact_order
```

```python
from dpone_airflow_pack import workload_ids_from_gitops_domain

workload_ids = workload_ids_from_gitops_domain(
    repo_root,
    "sales",
    group="daily",
)
```

The resolver only reads YAML files and validates that every group member exists
in the same domain catalog.

Inspect the current cache status:

```bash
dpone-airflow-pack-cache-status \
  --cache-dir /opt/airflow/dags/.dpone-cache/airflow
```

## Airflow execution policy

`airflow-pack.json` can carry a scheduler-side execution policy:

```json
{
  "airflow": {
    "execution": {
      "task_executor": "KubernetesExecutor",
      "deferrable": true,
      "on_finish_action": "delete_succeeded_pod",
      "get_logs": true,
      "logging_interval_seconds": 60
    }
  }
}
```

The lightweight provider maps it to KubernetesPodOperator kwargs
(`executor`, `deferrable`, `on_finish_action`, `get_logs`, and
`logging_interval`) and treats those fields as pack-owned. In deferrable mode,
`logging_interval_seconds` controls how often the task resumes to fetch fresh
pod logs. DAG-local overrides may still set DAG-owned values such as `retries`,
`pool`, or `in_cluster`.
