diff --git a/examples/bitcoin-tracker/AIRFLOW_INTEGRATION.md b/examples/bitcoin-tracker/AIRFLOW_INTEGRATION.md index 082651f..a59702a 100644 --- a/examples/bitcoin-tracker/AIRFLOW_INTEGRATION.md +++ b/examples/bitcoin-tracker/AIRFLOW_INTEGRATION.md @@ -200,6 +200,10 @@ sys.path.insert(0, os.path.abspath(os.path.dirname(__file__) + "/../..")) from ingest_bitcoin_prices import fetch_bitcoin_price, insert_to_bigquery +# Configuration from environment variables +GCP_PROJECT_ID = os.environ["GCP_PROJECT_ID"] # required: fail at parse time rather than +# targeting a placeholder project that does not exist + # Default arguments default_args = { "owner": "data-engineering", @@ -256,7 +260,7 @@ with DAG( """Insert price data to BigQuery.""" price_data = context["ti"].xcom_pull(task_ids="fetch_bitcoin_price") - project_id = os.environ.get("GCP_PROJECT_ID", "<>") + project_id = GCP_PROJECT_ID dataset = "crypto_data" table = "bitcoin_prices" @@ -286,7 +290,7 @@ with DAG( task_id="check_data_quality", sql=f""" SELECT COUNT(*) > 0 - FROM `{os.environ.get('GCP_PROJECT_ID', '<>')}.crypto_data.bitcoin_prices` + FROM `{GCP_PROJECT_ID}.crypto_data.bitcoin_prices` WHERE DATE(timestamp) = CURRENT_DATE() """, use_legacy_sql=False, @@ -297,7 +301,7 @@ with DAG( task_id="verify_transformations", sql=f""" SELECT COUNT(*) > 0 - FROM `{os.environ.get('GCP_PROJECT_ID', '<>')}.crypto_data.daily_price_summary` + FROM `{GCP_PROJECT_ID}.crypto_data.daily_price_summary` WHERE price_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY) """, use_legacy_sql=False, diff --git a/examples/bitcoin-tracker/airflow-quickstart.sh b/examples/bitcoin-tracker/airflow-quickstart.sh index 7ca6788..5eb079e 100755 --- a/examples/bitcoin-tracker/airflow-quickstart.sh +++ b/examples/bitcoin-tracker/airflow-quickstart.sh @@ -18,7 +18,12 @@ NC='\033[0m' # No Color # Configuration PROJECT_DIR=$(pwd) AIRFLOW_HOME=${AIRFLOW_HOME:-$HOME/airflow} -GCP_PROJECT_ID=${GCP_PROJECT_ID:-<>} +GCP_PROJECT_ID=${GCP_PROJECT_ID:-} +if [ -z "$GCP_PROJECT_ID" ]; then + echo "❌ Error: GCP_PROJECT_ID environment variable not set" + echo "Usage: export GCP_PROJECT_ID=your-project-id && ./airflow-quickstart.sh" + exit 1 +fi echo -e "${BLUE}Configuration:${NC}" echo " Project Directory: $PROJECT_DIR" diff --git a/examples/bitcoin-tracker/load_bitcoin_price_batch.py b/examples/bitcoin-tracker/load_bitcoin_price_batch.py index dd45a97..5a24be9 100644 --- a/examples/bitcoin-tracker/load_bitcoin_price_batch.py +++ b/examples/bitcoin-tracker/load_bitcoin_price_batch.py @@ -6,6 +6,7 @@ from google.cloud import bigquery from datetime import datetime import os +import sys import json import tempfile @@ -67,7 +68,17 @@ def load_to_bigquery_batch(row, project_id, dataset_id="crypto_data", table_id=" os.unlink(temp_file) if __name__ == "__main__": - project_id = os.getenv("GCP_PROJECT_ID", "<>") + # Get project ID from environment or command line + project_id = os.getenv("GCP_PROJECT_ID") + + if not project_id and len(sys.argv) > 1: + project_id = sys.argv[1] + + if not project_id: + print("❌ Error: GCP_PROJECT_ID environment variable not set") + print("Usage: python load_bitcoin_price_batch.py [PROJECT_ID]") + print(" or: export GCP_PROJECT_ID=your-project-id && python load_bitcoin_price_batch.py") + sys.exit(1) print(f"🚀 Fetching Bitcoin price...") price_data = fetch_bitcoin_price() diff --git a/examples/bitcoin-tracker/runtime/ingest_bitcoin_prices.py b/examples/bitcoin-tracker/runtime/ingest_bitcoin_prices.py index 8107182..8265138 100644 --- a/examples/bitcoin-tracker/runtime/ingest_bitcoin_prices.py +++ b/examples/bitcoin-tracker/runtime/ingest_bitcoin_prices.py @@ -12,6 +12,19 @@ def main(): """Fetch Bitcoin price from CoinGecko API and insert to BigQuery""" + # Resolve the target project before doing any work, so a missing + # GCP_PROJECT_ID fails here rather than silently targeting a placeholder. + project_id = os.getenv("GCP_PROJECT_ID") + + if not project_id and len(sys.argv) > 1: + project_id = sys.argv[1] + + if not project_id: + print("❌ Error: GCP_PROJECT_ID environment variable not set") + print("Usage: python ingest_bitcoin_prices.py [PROJECT_ID]") + print(" or: export GCP_PROJECT_ID=your-project-id && python ingest_bitcoin_prices.py") + sys.exit(1) + print("🚀 Starting Bitcoin price ingestion...") # Fetch from CoinGecko API (free tier, no auth required) @@ -47,7 +60,6 @@ def main(): } # Insert to BigQuery - project_id = os.getenv("GCP_PROJECT_ID", "<>") dataset_id = "crypto_data" table_id = "bitcoin_prices"