shubhamraj-git commented on code in PR #53352:
URL: https://github.com/apache/airflow/pull/53352#discussion_r2211368790
##########
airflow-core/src/airflow/example_dags/use_case/dynamic_stakeholder_dag_generator.py:
##########
@@ -0,0 +1,156 @@
+"""
+"Tailwind" - A virtual / non existing wind park energy company that powers a
farm of win-mills to
+produce clean energy. The company has a strong demand to ETL sensor data from
the windmills as well
+as need to act on data events when base data changes or contracts with
customers renew. The company
+values also the DEI rules and has sustainable targets for clean energy and CO2
reduction.
+
+Event driven case: New data is dropped on a file system that triggers data
processing for wind
+energy. Data is loaded and Asset events are generated.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.sdk import Asset, dag, task
+
+# Sample data in case we don't have a data source
+SAMPLE_DATA: list[dict[str, Any]] = [
+ {"name": "Alice", "stakeholder": "stakeholder_a", "region": "NA",
"amount_usd": 100},
+ {"name": "Bob", "stakeholder": "stakeholder_b", "region": "EU",
"amount_usd": 200},
+ {"name": "Charlie", "stakeholder": "stakeholder_a", "region": "NA",
"amount_usd": 150},
+ {"name": "Dana", "stakeholder": "stakeholder_c", "region": "APAC",
"amount_usd": 300},
+]
+
+# Config for the stakeholder dags
+CONFIG: list[dict[str, Any]] = [
+ {
+ "name": "stakeholder_a",
+ "filter_column": "stakeholder",
+ "filter_value": "stakeholder_a",
+ "data_source": {"type": "local"},
+ },
+ {
+ "name": "stakeholder_b",
+ "filter_column": "stakeholder",
+ "filter_value": "stakeholder_b",
+ "data_source": {
+ "type": "s3",
+ "bucket": "example-bucket",
+ "key": "stakeholder_data.csv",
+ },
+ },
+]
+
+DOC_MD_DAG = (
+ """
+ ## Stakeholder reporting dynamic dag
+
+ Loads and processes data for stakeholder. Shows dynamic DAGs, asset
tracking, and S3/local data
+ integration.
+
+ **References:**
+ - [Airflow Dag
Concepts](https://airflow.apache.org/docs/apache-airflow/stable/concepts/dags.html)
+ -
[S3Hook](https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/_api/airflow/providers/amazon/aws/hooks/s3/index.html)
+ - [Dynamic Dag
Generation](https://airflow.apache.org/docs/apache-airflow/stable/howto/dynamic-dag-generation.html)
+ -
[Asset](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/assets.html)
+ - [TaskFlow
API](https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html)
+ """
+)
+
+LOAD_DATA_DOC_MD = (
+"""
+Loads stakeholder data from S3 or local sample. Demonstrates use of
+[S3Hook](https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/_api/airflow/providers/amazon/aws/hooks/s3/index.html)
+for data ingestion.
+"""
+)
+
+TRANSFORM_AND_EMIT_DOC_MD = (
+"""
+Filter and transform stakeholder data, emits as
+[asset](https://airflow.apache.org/docs/apache-airflow/stable/data-assets.html).
+"""
+)
+
+S3_CONN_ID: str = "aws_default"
+
+for config in CONFIG:
+ dynamic_dag_id: str = f"use_case_{config['name']}_report"
+
+ def make_subdag(dag_config: dict[str, Any], dag_id: str) -> None:
+ """Create a stakeholder subdag for the given config."""
+ ds: dict[str, Any] = dag_config.get("data_source", {})
+ tags: list[str] = [
+ "Example",
+ "Storyline",
+ "DynamicDagGeneration",
+ "Asset",
+ ] + (["S3"] if ds.get("type") == "s3" else [])
+
+ import os
+
+ import pandas as pd
+
+ @dag(
+ dag_id=dag_id,
+ schedule="@monthly",
+ tags=tags,
+ description=(
+ f"Dynamically generated Dag to process data for stakeholder
{dag_config['name']}. "
+ "Demonstrates dynamic Dags, asset tracking, and S3/local data
integration."
+ ),
+ doc_md=DOC_MD_DAG,
+ )
+ def stakeholder_subdag() -> None:
Review Comment:
No, let me remove it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
