Metadata-Version: 2.4
Name: dagtool
Version: 0.0.1
Summary: a fast & friendly airflow dag factory
Project-URL: Homepage, https://github.com/ddeutils/dagtool/
Project-URL: Source Code, https://github.com/ddeutils/dagtool/
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: apache-airflow<2.8.0,>=2.7.1
Requires-Dist: jinja2==3.1.6
Requires-Dist: pendulum<3.0.0
Requires-Dist: pydantic
Requires-Dist: pyyaml==6.0.2
Provides-Extra: dev
Requires-Dist: clishelf>=0.2.22; extra == 'dev'
Requires-Dist: coverage>=7.10.0; extra == 'dev'
Requires-Dist: pre-commit>=4.0.0; extra == 'dev'
Requires-Dist: pytest>=8.4.0; extra == 'dev'
Provides-Extra: google
Requires-Dist: apache-airflow-providers-google; extra == 'google'
Description-Content-Type: text/markdown

# DAG Tool

A **Friendly Airflow DAG Build Tool** for Data Engineer with YAML file template.

> [!WARNING]
> This project will reference the DAG generate code from the [Astronomer: DAG-Factory](https://github.com/astronomer/dag-factory).
> But I replace some logic that fit with ETL propose for Data Engineer.

> [!NOTE]
> **Disclaimer**: This project will override all necessary parameters that should
> pass to Airflow object with Data Engineer context. So, if you want to use this
> project concept, you can enhance it with your idea.

**File Structure**:

```text
dags/
├── { domain }/
│     ├── { module-dags }/
│     │     ├── __init__.py
│     │     ├── dag.yml
│     │     ├── variables.yml
│     │     └── assets/
│     │         ├── dag-schema-mapping.json
│     │         └── dag-transform-query.sql
│     │
│     └── { module-dags }/
│           ├── __init__.py
```

> [!NOTE]
> I think this project should support multiple DAGs structure like:
>
> ```text
> dags/
> ├── { domain }/
> │     ├── { module-dags }/
> │     │     ├── __init__.py
> │     │     ├── dag-{ name-1 }.yml
> │     │     ├── dag-{ name-2 }.yml
> │     │     ├── variables.yml
> │     │     └── assets/
> │     │         ├── dag-case-1-schema-mapping.json
> │     │         ├── dag-case-1-transform-query.sql
> │     │         ├── dag-case-2-schema-mapping.json
> │     │         └── dag-case-2-transform-query.sql
> │     │
> │     └── { module-dags }/
> │           ├── __init__.py
> ```

**Execution Flow**:

The flow of this project provide the interface Pydantic Model before
passing it to Airflow objects.

```text
S --> Template --> Pydantic Model --> DAG/Operator Objects --> Execute --> E
```

**CI Flow**:

```text
S --> DAGs      --> GitSync     --> Airflow K8s Pod
  --> Variables --> API Sync    --> Airflow Variables
  --> Assets    --> CI Merge
```

**Feature Supported**:

- JSON Schema validation
- Passing environment variable
- Allow Passing Airflow Template

## 📦 Installation

> [!WARNING]
> This package does not publish to PyPI yet.

```shell
uv pip install -U dagtool
```

| Airflow Version  | Supported | Noted                                                          |
|:----------------:|:---------:|----------------------------------------------------------------|
|     `2.7.1`      |     ✅     | This is the first Airflow version that this project supported. |
| `>=2.7.1,<3.0.0` |     ❌     | Common version support for Airflow version `2.x.x`             |
|    `>=3.x.x`     |     ❌     | Common version support for Airflow version `3.x.x`             |

## 🎯 Usage

This DAG generator engine need you define the `dag.yml` file and set engine
object to get the current path on `__init__.py` file.

> [!NOTE]
> If you want to dynamic environment config on the `dag.yaml` file, you can use a
> `variable.yaml` file for dynamic value that marking on config template via macro
> function, `{{ var('keystore-on-dag-name') }}`.

```yaml
name: transaction
schedule: "@daily"
owner: "de-oncall@email.com,de@email.com"
authors: ["de-team"]
tags: ["sales", "tier-1", "daily"]
tasks:
  - task: start
    op: empty

  - group: etl_sales_master
    upstream: start
    tasks:
      - type: extract
        op: python
        uses: libs.gcs.csv@1.1.0
        assets:
          - name: schema-mapping.json
            alias: schema
            convertor: basic
        params:
          path: gcs://{{ var("PROJECT_ID") }}/sales/master/date/{ exec_date:%y }

      - task: transform
        upstream: extract
        op: docker
        uses: docker.rgt.co.th/image.transform:0.0.1
        assets:
          - name: transform-query.sql
            alias: transform
        params:
          path: gcs://{{ var("PROJECT_ID") }}/landing/master/date/{ exec_date:%y }

      - task: sink
        op: python
        run: |
          import time
          time.sleep(5)

  - task: end
    upstream: etl_sales_master
    op: empty
```

```python
"""# SALES DAG

This DAG will extract data from Google Cloud Storage to Google BigQuery LakeHouse
via DuckDB engine.

> This DAG is the temp DAG for ingest data to GCP.
"""
from dagtool import DagTool

dag = DagTool("sales", path=__file__, docs=__doc__)
dag.build_to_globals(gb=globals())
```

**Output**:

The DAG that was built from this package will have the name is, `sales_transaction`.

## 💬 Contribute

I do not think this project will go around the world because it has specific propose,
and you can create by your coding without this project dependency for long term
solution. So, on this time, you can open [the GitHub issue on this project :raised_hands:](https://github.com/ddeutils/dagtool/issues)
for fix bug or request new feature if you want it.
