kovagoadi/hunfabric
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve and fix DAG variable retrive method. (#3227)

Browse Source 
Improve and fix DAG variable retrive method in Airflow Demo DAG files.
This commit is contained in:
lcaggio
2025-07-10 10:21:11 +02:00
committed by GitHub
parent 88051c2d5b
commit 318da4c0c2
2 changed files with 12 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
fast/stages/3-data-platform-dev/demo/composer/DAG-dp0/gcs2bq_elt.py
View File

@@ -43,12 +43,12 @@ from airflow.utils.task_group import TaskGroup
LANDING_TABLES = ["users", "orders", "order_items", "products"]
# Environment variables (set from composer/variables.json)
DP_PROJECT = os.environ.get("DP_PROJECT")
LAND_BQ_DATASET = os.environ.get("LAND_BQ_DATASET")
CURATED_BQ_DATASET = os.environ.get("CURATED_BQ_DATASET")
LAND_GCS = os.environ.get("LAND_GCS")
DP_PROCESSING_SERVICE_ACCOUNT = os.environ.get("DP_PROCESSING_SERVICE_ACCOUNT")
LOCATION = os.environ.get("LOCATION")
DP_PROJECT = Variable.get("DP_PROJECT")
LAND_BQ_DATASET = Variable.get("LAND_BQ_DATASET")
CURATED_BQ_DATASET = Variable.get("CURATED_BQ_DATASET")
LAND_GCS = Variable.get("LAND_GCS")
DP_PROCESSING_SERVICE_ACCOUNT = Variable.get("DP_PROCESSING_SERVICE_ACCOUNT")
LOCATION = Variable.get("LOCATION")
# Validate required environment variables
required_vars = {

12
fast/stages/3-data-platform-dev/demo/composer/DAG-dp0/gcs2bq_table_create.py
View File

@@ -36,12 +36,12 @@ LANDING_TABLES = ["users", "orders", "order_items", "products"]
CURATED_TABLES = ["customer_purchases"]
# Environment variables (set from Composer variables.json)
DP_PROJECT = os.environ.get("DP_PROJECT")
LAND_BQ_DATASET = os.environ.get("LAND_BQ_DATASET")
CURATED_BQ_DATASET = os.environ.get("CURATED_BQ_DATASET")
EXPOSURE_BQ_DATASET = os.environ.get("EXPOSURE_BQ_DATASET")
LAND_GCS = os.environ.get("LAND_GCS")
DP_PROCESSING_SERVICE_ACCOUNT = os.environ.get("DP_PROCESSING_SERVICE_ACCOUNT")
DP_PROJECT = Variable.get("DP_PROJECT")
LAND_BQ_DATASET = Variable.get("LAND_BQ_DATASET")
CURATED_BQ_DATASET = Variable.get("CURATED_BQ_DATASET")
EXPOSURE_BQ_DATASET = Variable.get("EXPOSURE_BQ_DATASET")
LAND_GCS = Variable.get("LAND_GCS")
DP_PROCESSING_SERVICE_ACCOUNT = Variable.get("DP_PROCESSING_SERVICE_ACCOUNT")
# Validate required environment variables
required_vars = {