jscheffl commented on code in PR #53352:
URL: https://github.com/apache/airflow/pull/53352#discussion_r2205836853
##########
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",
Review Comment:
Could we invent just two names instead of "a" and "b"? Maybe just artificial
names like "miller" or "smith"?
##########
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:
Is pandas in default Airflow dependencies? Or would we need to wrap this
into a virtualenv to ensure it is loadable in a default setup?
##########
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:
Typing is cool! But do we need to define `-> None` on the dag level? Is this
common best practice? Not sure.
##########
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:
Review Comment:
Why do you add a function to generate the Dag? Is this a best practice? Not
sure, whenever I did it I just started here with the `@dag` inside the loop and
generated it. But no strong opinionn
##########
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:
+ @task(doc_md=LOAD_DATA_DOC_MD)
Review Comment:
And also a good idea... actually I considered one dag where we loop to
generate different tasks based on config... but a dag per config is also fine.
##########
airflow-core/src/airflow/example_dags/use_case/dynamic_stakeholder_dag_generator.py:
##########
Review Comment:
I think the name of the dag should be according to the use case, not what
technically is produced.
##########
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",
Review Comment:
Could we use local file and drop the CSV besides the dag in the examples
folder?
##########
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
Review Comment:
Example must be w/o other provider dependency. Might be extended by an
example in Amazon provider. But import must be removed here :-(
##########
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:
+ @task(doc_md=LOAD_DATA_DOC_MD)
Review Comment:
As we might take code snippets in the documentation, probably we shoudl have
the doc_md inline?
--
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]
