DAG Tool

Warning

This project will reference the DAG generate code from the 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 ETL context for Data Engineer use-case. So, if you want to use and enhance this project, you can fork this project anytime without notice me.

Note

I recommend to use Airflow2 until Airflow3 stable.

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:

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

📦 Installation

uv pip install -U common

📍 Usage

DAG Template

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, {{ vars('keystore-on-dag-name') }}.

name: transaction
schedule: "@daily"
owner: "de-oncall@email.com,de@email.com"
start_date: "{{ vars('start_date') }}"
catchup: "{{ vars('catchup') }}"
tags:
 - "domain:sales"
 - "tier:1"
 - "schedule:daily"
tasks:
 - task: start
 op: empty
 - group: etl_master
 upstream: start
 tasks:
 - type: extract
 op: python
 caller: get_api_data
 params:
 path: gcs://{{ vars("project_id") }}/sales/master/date/{{ exec_date | fmt('%y') }}
 - task: transform
 upstream: extract
 op: operator
 operator_name: gcs_transform_data
 params:
 path: gcs://{{ vars("project_id") }}/landing/master/date/{{ exec_date | fmt('%y') }}
 - task: sink
 upstream: transform
 op: common
 uses: write_iceberg
 params:
 path: gcs://{{ vars("project_id") }}
 - task: end
 upstream: etl_master
 op: empty

"""# 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 Factory, ToolModel, BuildContext
from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.utils.task_group import TaskGroup
from airflow.operators.empty import EmptyOperator
# NOTE: Some external provider operator object.
from airflow.providers.google.cloud.operators import MockGCSTransformData
from pydantic import Field
# NOTE: Some function that want to use with PythonOperator
def get_api_data(path: str) -> dict[str, list[str]]:
 return {"data": [f"src://{path}/table/1", f"src://{path}/table/2"]}
# NOTE: Some common task that create any Airflow Task instance object.
class WriteIcebergTool(ToolModel):
 """Custom Task for user defined inside of template path."""
 path: str = Field(description="An Iceberg path.")
 def build(
 self,
 dag: DAG,
 task_group: TaskGroup | None = None,
 build_context: BuildContext | None = None,
 ) -> TaskGroup:
 with TaskGroup(
 group_id="write_iceberg",
 parent_group=task_group,
 dag=dag,
 ) as tg:
 t1 = EmptyOperator(task_id="prepare", dag=dag)
 t2 = EmptyOperator(task_id="write", dag=dag)
 t2.set_upstream(t1)
 return tg
factory = Factory(
 name="sales",
 path=__file__,
 docs=__doc__,
 operators={"gcs_transform_data": MockGCSTransformData},
 python_callers={"get_api_data": get_api_data},
 tools={"write_iceberg": WriteIcebergTool},
)
factory.build_airflow_dags_to_globals(
 gb=globals(),
 default_args={"start_date": days_ago(2)},
)

Note

On the variables.yml file that will set different stage area variables:

type: variable
variables:
 # NOTE: The key name that will get from the Airflow Variable or the local
 # variable file.
 - key: transaction
 stages:
 dev:
 start_date: "2025年01月01日"
 catchup: false
 project_id: "sales_project_dev"
 prod:
 start_date: "2025年01月31日"
 catchup: true
 project_id: "sales_project"

🎯 Roadmaps

💬 Contribute