69 - Capstone Project Implementation Part 2

SELECT
    SHA256(t.transaction_id) AS transaction_id,
    t.product,
    t.price,
    t.quantity,
    CAST(t.date AS DATE) AS sale_date
FROM {{ ref('stg_transactions') }} t

from airflow import DAG
from airflow.operators.python import PythonOperator
with DAG(dag_id="transform_sales_mart", schedule_interval="@daily"):
    run_dbt = PythonOperator(
        task_id="run_dbt",
        python_callable=run_dbt_multi_target,
    )

workers:
  resources:
    limits:
      cpu: '500m'
      memory: '1Gi'

import pytest
from google.cloud import bigquery

@pytest.fixture
def client():
    return bigquery.Client()

def test_fact_sales(client):
    result = list(client.query("SELECT COUNT(*) AS cnt FROM `project.dataset.fact_sales`").result())
    assert result[0]["cnt"] == 6000

pipeline:
  name: transaction_pipeline
  version: 1.0.0
data_lake:
  gcs_bucket: hijra-transactions
  path: raw/transactions
data_warehouse:
  project_id: hijra-project
  dataset: transactions_warehouse
  table: raw_transactions
  postgres:
    secretName: postgres-credentials
    dbname: mydb
    host: localhost
    port: 5432
data_mart:
  dataset: transactions_mart
  table: sales_by_product
ingestion:
  fastapi_endpoint: /upload/transactions
  batch_size: 10000
  trust_input: true
  validation:
    min_price: 10.0
    max_quantity: 100
    required_fields:
      - transaction_id
      - product
      - price
      - quantity
      - date
    product_prefix: 'Halal'
    max_decimals: 2
transformation:
  dbt_project: dbt_transactions
  models:
    - sales_by_product
    - sales_by_date
orchestration:
  airflow_dag: transaction_etl
  schedule: '0 0 * * *'
security:
  pii_fields:
    - transaction_id
kubernetes:
  namespace: data-pipeline
  helm_chart: transaction-pipeline
  resources:
    cpu: '1'
    memory: '2Gi'

# File: de-onboarding/utils.py
from typing import Dict, Any, Union
import yaml
import pandas as pd
from kubernetes import client, config

def read_config(config_path: str) -> Dict[str, Any]:
    """Read YAML configuration."""
    with open(config_path, "r") as file:
        config: Dict[str, Any] = yaml.safe_load(file)
    print(f"Loaded config: {config}")
    return config

def read_k8s_secret(secret_name: str, namespace: str = "data-pipeline") -> Dict[str, str]:
    """Read Kubernetes secret."""
    config.load_kube_config()
    v1 = client.CoreV1Api()
    secret = v1.read_namespaced_secret(secret_name, namespace)
    return {k: v.decode('utf-8') for k, v in secret.data.items()}

def is_numeric(s: str, max_decimals: int = 2) -> bool:
    """Check if string is a decimal number."""
    parts = s.split(".")
    if len(parts) != 2 or not parts[0].replace("-", "").isdigit() or not parts[1].isdigit():
        return False
    return len(parts[1]) <= max_decimals

def clean_string(s: Union[str, None]) -> str:
    """Strip whitespace from string."""
    return s.strip() if s else ""

def is_numeric_value(x: Any) -> bool:
    """Check if value is numeric."""
    return isinstance(x, (int, float))

def is_integer(x: Any) -> bool:
    """Check if value is an integer."""
    return str(x).isdigit()

def validate_sale(sale: Dict[str, Any], config: Dict[str, Any]) -> bool:
    """Validate sale based on config rules."""
    required_fields = config["required_fields"]
    min_price = config["min_price"]
    max_quantity = config["max_quantity"]
    prefix = config["product_prefix"]
    max_decimals = config["max_decimals"]

    print(f"Validating sale: {sale}")
    for field in required_fields:
        if field not in sale or not sale[field] or clean_string(str(sale[field])) == "":
            print(f"Invalid sale: missing {field}: {sale}")
            return False

    product = clean_string(sale["product"])
    if not product.startswith(prefix):
        print(f"Invalid sale: product lacks '{prefix}' prefix: {sale}")
        return False

    price = clean_string(str(sale["price"]))
    if not is_numeric(price, max_decimals) or float(price) < min_price or float(price) <= 0:
        print(f"Invalid sale: invalid price: {sale}")
        return False

    quantity = clean_string(str(sale["quantity"]))
    if not is_integer(quantity) or int(quantity) > max_quantity:
        print(f"Invalid sale: invalid quantity: {sale}")
        return False

    return True

# File: de-onboarding/preprocess.py
from typing import Tuple
import pandas as pd
import utils
from google.cloud import bigquery
import psycopg2

def preprocess_transactions(csv_path: str, config_path: str, bq_table_id: str) -> Tuple[pd.DataFrame, int, int]:
    """Validate and upload transactions to BigQuery and PostgreSQL."""
    config = utils.read_config(config_path)
    trust_input = config.get("ingestion", {}).get("trust_input", False)
    validation_config = config["ingestion"]["validation"]
    pg_config = config["data_warehouse"]["postgres"]
    secret_name = pg_config.get("secretName")

    pg_conn_params = {
        "dbname": pg_config["dbname"],
        "host": pg_config["host"],
        "port": pg_config["port"]
    }
    if secret_name:
        secret_data = utils.read_k8s_secret(secret_name)
        pg_conn_params["user"] = secret_data["user"]
        pg_conn_params["password"] = secret_data["password"]
    else:
        pg_conn_params["user"] = pg_config["user"]
        pg_conn_params["password"] = pg_config["password"]

    df = pd.read_csv(csv_path)
    print(f"Initial DataFrame:\n{df.head()}")

    total_records = len(df)
    if trust_input:
        print("Trusting input from Chapter 68; skipping validation")
        valid_df = df
    else:
        valid_rows = [row for _, row in df.iterrows() if utils.validate_sale(row.to_dict(), validation_config)]
        if not valid_rows:
            print("No valid transactions")
            return pd.DataFrame(), 0, total_records
        valid_df = pd.DataFrame(valid_rows)
        print(f"Validated DataFrame:\n{valid_df}")

    client = bigquery.Client()
    valid_df.to_gbq(bq_table_id, if_exists="replace")
    conn = psycopg2.connect(**pg_conn_params)
    valid_df.to_sql("transactions", conn, if_exists="replace", index=False)
    conn.close()

    return valid_df, len(valid_df), total_records

# File: de-onboarding/run_dbt_multi_target.py
from typing import None
import subprocess
import utils

def run_dbt_multi_target(config_path: str) -> None:
    """Run dbt for BigQuery and PostgreSQL targets."""
    config = utils.read_config(config_path)
    dbt_project_dir = config["transformation"]["dbt_project"]

    for target in ["bigquery", "postgres"]:
        print(f"Running dbt for target: {target}")
        result = subprocess.run(
            ["dbt", "run", "--profiles-dir", ".", "--profile", target],
            cwd=dbt_project_dir,
            capture_output=True,
            text=True
        )
        if result.returncode != 0:
            print(f"dbt run failed for {target}: {result.stderr}")
            raise RuntimeError(f"dbt run failed for {target}")
        print(f"dbt run completed for {target}: {result.stdout}")

-- File: de-onboarding/dbt/models/mart/fact_sales.sql
{{
    config(
        materialized='table',
        partition_by={
            "field": "sale_date",
            "data_type": "date"
        }
    )
}}

SELECT
    SHA256(t.transaction_id) AS transaction_id,
    t.product,
    t.price,
    t.quantity,
    t.price * t.quantity AS amount,
    CAST(t.date AS DATE) AS sale_date
FROM {{ ref('stg_transactions') }} t
WHERE t.product IS NOT NULL
  AND t.product LIKE 'Halal%'
  AND t.price > 0
  AND t.quantity <= 100

-- File: de-onboarding/dbt/models/mart/summary_sales.sql
{{
    config(
        materialized='view'
    )
}}

SELECT
    product,
    DATE_TRUNC('month', CAST(sale_date AS DATE)) AS sale_month, -- Aligns with Chapter 67's sales_by_date
    SUM(price * quantity) AS total_amount,
    AVG(price) AS avg_price,
    COUNT(DISTINCT transaction_id) AS transaction_count
FROM {{ ref('fact_sales') }}
GROUP BY product, sale_month

-- File: de-onboarding/dbt/models/mart/pg_sales.sql
{{
    config(
        materialized='table'
    )
}}

SELECT
    t.transaction_id AS transaction_id, -- Hashed from Chapter 68
    t.product,
    t.price,
    t.quantity,
    t.price * t.quantity AS amount,
    t.date AS sale_date
FROM {{ source('postgres', 'transactions') }} t
WHERE t.product IS NOT NULL
  AND t.product LIKE 'Halal%'
  AND t.price > 0
  AND t.quantity <= 100;

CREATE INDEX IF NOT EXISTS idx_sale_date ON {{ this }} (sale_date);

-- File: de-onboarding/dbt/models/staging/stg_transactions.sql
SELECT
    transaction_id,
    product,
    price,
    quantity,
    date
FROM {{ source('raw', 'transactions') }}

# File: de-onboarding/export_summary.py
from typing import None
from google.cloud import bigquery
import json
import pandas as pd

def export_summary(table_id: str, json_path: str, csv_path: str) -> None:
    """Export summary_sales to JSON and CSV."""
    client = bigquery.Client()
    query = f"SELECT * FROM `{table_id}`"
    df = client.query(query).to_dataframe()
    print(f"Summary DataFrame:\n{df}")
    with open(json_path, "w") as f:
        json.dump(df.to_dict(orient="records"), f, indent=2)
    df.to_csv(csv_path, index=False)
    print(f"Exported summary to {json_path} and {csv_path}")

# File: de-onboarding/generate_large_data.py
from typing import None
import pandas as pd
import numpy as np

def generate_large_transactions(output_path: str, num_rows: int = 10000) -> None:
    """Generate large transaction dataset using NumPy (Chapter 3)."""
    products = ["Halal Laptop", "Halal Mouse", "Halal Keyboard", "Monitor", ""]
    dates = pd.date_range("2023-10-01", "2023-10-10").strftime("%Y-%m-%d")

    df = pd.DataFrame({
        "transaction_id": [f"T{i:04d}" for i in range(1, num_rows + 1)],
        "product": np.random.choice(products, num_rows, p=[0.3, 0.3, 0.3, 0.05, 0.05]),
        "price": np.random.uniform(10, 1000, num_rows).round(2),
        "quantity": np.random.randint(1, 150, num_rows),
        "date": np.random.choice(dates, num_rows)
    })
    df.to_csv(output_path, index=False)
    print(f"Generated {output_path} with {num_rows} rows")

# File: de-onboarding/airflow/dags/transform_dag.py
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from typing import Any
import preprocess
import export_summary
import run_dbt_multi_target
import utils

default_args: dict[str, Any] = {
    "owner": "hijra_group",
    "retries": 3,
    "retry_delay": 300,
}

with DAG(
    dag_id="transform_sales_mart",
    start_date=datetime(2023, 10, 1),
    schedule_interval="@daily",
    default_args=default_args,
    catchup=False,
) as dag:
    config = utils.read_config("data/pipeline_config.yaml")

    preprocess_task = PythonOperator(
        task_id="preprocess_transactions",
        python_callable=preprocess.preprocess_transactions,
        op_kwargs={
            "csv_path": "data/large_transactions.csv",
            "config_path": "data/pipeline_config.yaml",
            "bq_table_id": "project.dataset.transactions"
        }
    )

    run_dbt = PythonOperator(
        task_id="run_dbt",
        python_callable=run_dbt_multi_target.run_dbt_multi_target,
        op_kwargs={"config_path": "data/pipeline_config.yaml"}
    )

    export_summary_task = PythonOperator(
        task_id="export_summary",
        python_callable=export_summary.export_summary,
        op_kwargs={
            "table_id": "project.dataset.summary_sales",
            "json_path": "data/summary_sales.json",
            "csv_path": "data/summary_sales.csv"
        }
    )

    preprocess_task >> run_dbt >> export_summary_task

# File: de-onboarding/helm/airflow/values.yaml
executor: KubernetesExecutor
workers:
  resources:
    limits:
      cpu: '500m'
      memory: '1Gi'
    requests:
      cpu: '200m'
      memory: '512Mi'
secrets:
  - name: bigquery-credentials
    key: credentials.json
  - name: postgres-credentials
    key: credentials.yaml

# File: de-onboarding/tests/test_mart.py
from google.cloud import bigquery
import pytest
import json
import os
import psycopg2
from typing import List

@pytest.fixture
def bq_client() -> bigquery.Client:
    return bigquery.Client()

@pytest.fixture
def pg_conn():
    conn = psycopg2.connect(
        dbname="mydb", user="myuser", password="mypassword", host="localhost", port="5432"
    )
    yield conn
    conn.close()

def test_fact_sales(bq_client: bigquery.Client):
    """Test BigQuery data mart row count."""
    query = "SELECT COUNT(*) AS cnt FROM `project.dataset.fact_sales`"
    result = list(bq_client.query(query).result())
    assert result[0]["cnt"] == 6000, "Expected 6000 valid sales records"

def test_pg_sales(pg_conn):
    """Test PostgreSQL data mart row count, relying on Chapter 68's hashed transaction_id."""
    cursor = pg_conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM pg_sales")
    result = cursor.fetchone()
    assert result[0] == 6000, "Expected 6000 valid sales records in PostgreSQL"

def test_performance(bq_client: bigquery.Client):
    """Test BigQuery query performance."""
    query = """
    SELECT SUM(total_amount)
    FROM `project.dataset.summary_sales`
    WHERE sale_month = '2023-10-01'
    """
    job = bq_client.query(query)
    result = list(job.result())
    assert job.total_bytes_processed < 10 * 1024 * 1024, "Query too costly"

def test_summary_output():
    """Test summary file contents."""
    for path in ["data/summary_sales.json", "data/summary_sales.csv"]:
        assert os.path.exists(path), f"Summary file {path} not created"
    with open("data/summary_sales.json", "r") as f:
        summary = json.load(f)
    assert len(summary) > 0, "Summary is empty"
    assert all(
        "product" in row and "sale_month" in row and "total_amount" in row and "avg_price" in row and "transaction_count" in row
        for row in summary
    ), "Invalid summary format"

# File: de-onboarding/tests/test_pii.py
from google.cloud import bigquery
import pytest
from typing import List

@pytest.fixture
def client() -> bigquery.Client:
    return bigquery.Client()

def test_pii_masking(client: bigquery.Client):
    """Test PII hashing in BigQuery data mart."""
    query = "SELECT transaction_id FROM `project.dataset.fact_sales`"
    result = list(client.query(query).result())
    for row in result:
        assert len(row["transaction_id"]) == 64, "Transaction ID not hashed"

# File: de-onboarding/tests/test_secrets.py
import pytest
from unittest.mock import patch
import utils

def test_read_k8s_secret():
    """Test reading Kubernetes secret."""
    with patch("kubernetes.client.CoreV1Api.read_namespaced_secret") as mock_secret:
        mock_secret.return_value.data = {
            "user": "myuser".encode('utf-8'),
            "password": "mypassword".encode('utf-8')
        }
        secret_data = utils.read_k8s_secret("postgres-credentials", "data-pipeline")
        assert secret_data == {"user": "myuser", "password": "mypassword"}

[
  {
    "product": "Halal Laptop",
    "sale_month": "2023-10-01",
    "total_amount": 1999.98,
    "avg_price": 999.99,
    "transaction_count": 1
  },
  {
    "product": "Halal Mouse",
    "sale_month": "2023-10-01",
    "total_amount": 249.9,
    "avg_price": 24.99,
    "transaction_count": 1
  }
]

transaction_id,product,price,quantity,amount,sale_date,total_amount
<hash>,Halal Laptop,999.99,2,1999.98,2023-10-01,<large_sum>

import subprocess
subprocess.run(["kubectl", "scale", "deployment", "airflow-worker", "--replicas=2", "--namespace=data-pipeline"])

run_dbt = BashOperator(
    task_id="run_dbt",
    bash_command="dbt run",
)

data_warehouse:
  postgres:
    secretName: wrong-secret
    dbname: mydb
    host: localhost
    port: 5432

-- File: de-onboarding/dbt/models/mart/fact_sales.sql
{{
    config(
        materialized='table',
        partition_by={
            "field": "sale_date",
            "data_type": "date"
        }
    )
}}

SELECT
    SHA256(t.transaction_id) AS transaction_id,
    t.product,
    t.price,
    t.quantity,
    t.price * t.quantity AS amount,
    CAST(t.date AS DATE) AS sale_date,
    SUM(t.price * t.quantity) OVER (PARTITION BY t.product) AS total_amount
FROM {{ ref('stg_transactions') }} t
WHERE t.product IS NOT NULL
  AND t.product LIKE 'Halal%'
  AND t.price > 0
  AND t.quantity <= 100

# File: de-onboarding/airflow/dags/transform_dag.py
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from typing import Any
import preprocess
import export_summary
import run_dbt_multi_target
import utils

default_args: dict[str, Any] = {
    "owner": "hijra_group",
    "retries": 5,
    "retry_delay": 300,
}

with DAG(
    dag_id="transform_sales_mart",
    start_date=datetime(2023, 10, 1),
    schedule_interval="@daily",
    default_args=default_args,
    catchup=False,
) as dag:
    config = utils.read_config("data/pipeline_config.yaml")

    preprocess_task = PythonOperator(
        task_id="preprocess_transactions",
        python_callable=preprocess.preprocess_transactions,
        op_kwargs={
            "csv_path": "data/large_transactions.csv",
            "config_path": "data/pipeline_config.yaml",
            "bq_table_id": "project.dataset.transactions"
        }
    )

    run_dbt = PythonOperator(
        task_id="run_dbt",
        python_callable=run_dbt_multi_target.run_dbt_multi_target,
        op_kwargs={"config_path": "data/pipeline_config.yaml"}
    )

    export_summary_task = PythonOperator(
        task_id="export_summary",
        python_callable=export_summary.export_summary,
        op_kwargs={
            "table_id": "project.dataset.summary_sales",
            "json_path": "data/summary_sales.json",
            "csv_path": "data/summary_sales.csv"
        }
    )

    preprocess_task >> run_dbt >> export_summary_task

# File: de-onboarding/scalability_test.py
from google.cloud import bigquery
import pandas as pd
import time
import subprocess
from typing import None

def test_scalability() -> None:
    client = bigquery.Client()
    df = pd.DataFrame({
        "transaction_id": [f"T{i:04d}" for i in range(10000)],
        "product": ["Halal Laptop"] * 10000,
        "price": [999.99] * 10000,
        "quantity": [2] * 10000,
        "date": ["2023-10-01"] * 10000
    })
    df.to_gbq("project.dataset.transactions", if_exists="replace")

    subprocess.run(["kubectl", "scale", "deployment", "airflow-worker", "--replicas=2", "--namespace=data-pipeline"])

    start = time.time()
    client.query("SELECT * FROM `project.dataset.fact_sales`").result()
    runtime = time.time() - start
    assert runtime < 10, f"Runtime {runtime}s exceeds 10s"
    print("Conceptual Answer: More pods parallelize dbt tasks, reducing runtime by ~30% via Kubernetes resource allocation.")

# File: de-onboarding/tests/test_pii.py
from google.cloud import bigquery
import pytest
from typing import List

@pytest.fixture
def client() -> bigquery.Client:
    return bigquery.Client()

def test_pii_masking(client: bigquery.Client):
    query = "SELECT transaction_id FROM `project.dataset.fact_sales`"
    result = list(client.query(query).result())
    for row in result:
        assert len(row["transaction_id"]) == 64, "Transaction ID not hashed"

# File: de-onboarding/airflow/dags/transform_dag.py
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from typing import Any

default_args: dict[str, Any] = {
    "owner": "hijra_group",
    "retries": 3,
    "retry_delay": 300,
}

with DAG(
    dag_id="transform_sales_mart",
    start_date=datetime(2023, 10, 1),
    schedule_interval="@daily",
    default_args=default_args,
    catchup=False,
) as dag:
    run_dbt = PythonOperator(
        task_id="run_dbt",
        python_callable=run_dbt_multi_target.run_dbt_multi_target,
        op_kwargs={"config_path": "data/pipeline_config.yaml"}
    )

    run_dbt

# File: de-onboarding/debug_postgres.py
from typing import Dict, Any
import psycopg2
import yaml

def test_postgres_connection(config_path: str) -> None:
    """Test PostgreSQL connection with debug logging."""
    config = yaml.safe_load(open(config_path))
    pg_config = config["data_warehouse"]["postgres"]
    secret_name = pg_config.get("secretName")

    conn_params = {
        "dbname": pg_config["dbname"],
        "host": pg_config["host"],
        "port": pg_config["port"]
    }
    if secret_name:
        secret_data = utils.read_k8s_secret(secret_name)
        conn_params["user"] = secret_data["user"]
        conn_params["password"] = secret_data["password"]
    else:
        conn_params["user"] = pg_config["user"]
        conn_params["password"] = pg_config["password"]

    print(f"Attempting connection with params: {conn_params}")
    try:
        conn = psycopg2.connect(**conn_params)
        print("Connection successful")
        conn.close()
    except Exception as e:
        print(f"Connection failed: {str(e)}")
        raise

# Fix by setting secretName='postgres-credentials' in pipeline_config.yaml
if __name__ == "__main__":
    test_postgres_connection("data/pipeline_config.yaml")