shubhamraj-git commented on code in PR #53352:
URL: https://github.com/apache/airflow/pull/53352#discussion_r2211359785
##########
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
Review Comment:
I think yes we need to figure this out, since with dags having use cases we
might need some more dependency.
--
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]
