70 - Capstone Project Implementation Part 3

# File: de-onboarding/app/main.py
from typing import List, Dict, Any
from fastapi import FastAPI, Depends, HTTPException, status, Request
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field
from google.cloud import bigquery
from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
import yaml
import hashlib
import logging
import time
import re
import psycopg2
import utils  # From Chapter 69

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Transaction Pipeline API")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
templates = Jinja2Templates(directory="templates")

# Jinja2 filter for PII masking, complementing Chapter 69's transaction_id hashing
def hash_product(value: str) -> str:
    """Hash product name for PII protection, aligning with Chapter 69's SHA-256 hashing."""
    if not value:
        return ""
    return hashlib.sha256(value.encode()).hexdigest()[:8]  # Shortened for display, consistent with Chapter 65

templates.env.filters["hash_product"] = hash_product

# Security setup
config = utils.read_config("data/pipeline_config.yaml")
SECRET_KEY = config.get("security", {}).get("jwt_secret", "your-secret-key")  # From pipeline_config.yaml
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# PostgreSQL connection
pg_config = config["data_warehouse"]["postgres"]
pg_conn_params = {
    "dbname": pg_config["dbname"],
    "host": pg_config["host"],
    "port": pg_config["port"],
    "user": pg_config.get("user", utils.read_k8s_secret(pg_config["secretName"], "data-pipeline")["user"]),
    "password": pg_config.get("password", utils.read_k8s_secret(pg_config["secretName"], "data-pipeline")["password"])
}

class User(BaseModel):
    username: str
    hashed_password: str

class Token(BaseModel):
    access_token: str
    token_type: str

class TransactionSummary(BaseModel):
    transaction_id: str
    product: str
    sale_month: str
    total_amount: float
    avg_price: float
    transaction_count: int

class QueryParams(BaseModel):
    product_prefix: str = Field(default="Halal", pattern=r"^[a-zA-Z0-9]+$")
    min_amount: float = Field(default=0.0, ge=0.0)
    max_quantity: int = Field(default=1000, ge=0)
    sale_month: str = Field(default="2023-10-01", pattern=r"^\d{4}-\d{2}-\d{2}$")

# Mock user database (in production, use PostgreSQL)
users_db = {
    "admin": {
        "username": "admin",
        "hashed_password": pwd_context.hash("password123")  # Hashed password
    }
}

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify password against hashed password."""
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: Dict[str, Any]) -> str:
    """Create JWT access token."""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> Dict[str, Any]:
    """Get current user from JWT token."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
        user = users_db.get(username)
        if user is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
        return user
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()) -> Token:
    """Authenticate user and return JWT token."""
    user = users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect credentials")
    access_token = create_access_token(data={"sub": user["username"]})
    logger.info(f"User {user['username']} logged in")
    return Token(access_token=access_token, token_type="bearer")

# BigQuery client
bq_client = bigquery.Client()

@app.get("/sales/trends", response_model=List[TransactionSummary])
async def get_sales_trends(
    params: QueryParams = Depends(),
    current_user: Dict[str, Any] = Depends(get_current_user)
) -> List[TransactionSummary]:
    """Get sales trends from BigQuery summary_sales view."""
    start_time = time.time()
    logger.info(f"User {current_user['username']} queried trends: product_prefix={params.product_prefix}, min_amount={params.min_amount}, max_quantity={params.max_quantity}, sale_month={params.sale_month}")
    try:
        query = """
        SELECT
            transaction_id,
            product,
            sale_month,
            total_amount,
            avg_price,
            transaction_count
        FROM `hijra-project.transactions_mart.summary_sales`
        WHERE product LIKE @prefix || '%'
        AND sale_month = @sale_month
        """
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                {"name": "prefix", "parameterType": {"type": "STRING"}, "parameterValue": {"value": params.product_prefix}},
                {"name": "sale_month", "parameterType": {"type": "STRING"}, "parameterValue": {"value": params.sale_month}}
            ]
        )
        logger.info(f"Executing BigQuery summary query")
        query_job = bq_client.query(query, job_config=job_config)
        results = [
            TransactionSummary(
                transaction_id=row["transaction_id"],  # Hashed from Chapter 69
                product=row["product"],
                sale_month=row["sale_month"],
                total_amount=row["total_amount"],
                avg_price=row["avg_price"],
                transaction_count=row["transaction_count"]
            ) for row in query_job
        ]
        logger.info(f"Retrieved {len(results)} sales summaries")
        return results
    except Exception as e:
        logger.error(f"BigQuery query failed: {str(e)}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Query error: {str(e)}")
    finally:
        duration = time.time() - start_time
        logger.info(f"API response time: {duration:.2f} seconds")

@app.get("/sales/pg-trends", response_model=List[TransactionSummary])
async def get_pg_sales_trends(
    params: QueryParams = Depends(),
    current_user: Dict[str, Any] = Depends(get_current_user)
) -> List[TransactionSummary]:
    """Get sales trends from PostgreSQL pg_sales table."""
    start_time = time.time()
    logger.info(f"User {current_user['username']} queried PostgreSQL trends: product_prefix={params.product_prefix}, min_amount={params.min_amount}, max_quantity={params.max_quantity}, sale_month={params.sale_month}")
    try:
        conn = psycopg2.connect(**pg_conn_params)
        cursor = conn.cursor()
        query = """
        SELECT
            transaction_id,
            product,
            TO_CHAR(sale_date, 'YYYY-MM-DD') AS sale_month,
            SUM(amount) AS total_amount,
            AVG(price) AS avg_price,
            COUNT(DISTINCT transaction_id) AS transaction_count
        FROM pg_sales
        WHERE product LIKE %s || '%%'
        AND TO_CHAR(sale_date, 'YYYY-MM-DD') = %s
        GROUP BY transaction_id, product, TO_CHAR(sale_date, 'YYYY-MM-DD')
        HAVING SUM(amount) >= %s AND SUM(quantity) <= %s
        """
        cursor.execute(query, (params.product_prefix, params.sale_month, params.min_amount, params.max_quantity))
        results = [
            TransactionSummary(
                transaction_id=row[0],
                product=row[1],
                sale_month=row[2],
                total_amount=row[3],
                avg_price=row[4],
                transaction_count=row[5]
            ) for row in cursor.fetchall()
        ]
        cursor.close()
        conn.close()
        logger.info(f"Retrieved {len(results)} PostgreSQL sales summaries")
        return results
    except Exception as e:
        logger.error(f"PostgreSQL query failed: {str(e)}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Query error: {str(e)}")
    finally:
        duration = time.time() - start_time
        logger.info(f"API response time: {duration:.2f} seconds")

@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request, current_user: Dict[str, Any] = Depends(get_current_user)) -> HTMLResponse:
    """Render sales trends dashboard."""
    start_time = time.time()
    summaries = await get_sales_trends(QueryParams(product_prefix="Halal", min_amount=0.0, max_quantity=1000, sale_month="2023-10-01"), current_user)
    logger.info("Rendering dashboard")
    response = templates.TemplateResponse(
        "dashboard.html",
        {"request": request, "summaries": summaries}
    )
    duration = time.time() - start_time
    logger.info(f"Dashboard render time: {duration:.2f} seconds")
    return response

<!-- File: de-onboarding/app/templates/dashboard.html -->
<!DOCTYPE html>
<html>
  <head>
    <title>Sales Trends Dashboard</title>
    <style>
      body {
        font-family: Arial, sans-serif;
        margin: 20px;
      }
      table {
        border-collapse: collapse;
        width: 100%;
      }
      th,
      td {
        border: 1px solid #ddd;
        padding: 8px;
        text-align: left;
      }
      th {
        background-color: #f2f2f2;
      }
    </style>
  </head>
  <body>
    <h1>Sales Trends Dashboard</h1>
    <table>
      <tr>
        <th>Product (Hashed)</th>
        <th>Sale Month</th>
        <th>Total Amount ($)</th>
        <th>Average Price ($)</th>
        <th>Transaction Count</th>
      </tr>
      {% for summary in summaries %}
      <tr>
        <td>{{ summary.product | hash_product }}</td>
        <td>{{ summary.sale_month }}</td>
        <td>{{ summary.total_amount | round(2) }}</td>
        <td>{{ summary.avg_price | round(2) }}</td>
        <td>{{ summary.transaction_count }}</td>
      </tr>
      {% endfor %}
    </table>
  </body>
</html>

# File: de-onboarding/helm/transaction-pipeline/Chart.yaml
apiVersion: v2
name: transaction-pipeline
description: Helm chart for transaction pipeline
version: 0.1.0

# File: de-onboarding/helm/transaction-pipeline/values.yaml
replicaCount: 3 # Support scaling

image:
  repository: transaction-pipeline
  tag: latest
  pullPolicy: IfNotPresent

service:
  type: ClusterIP
  port: 80

resources:
  limits:
    cpu: '500m'
    memory: '1Gi'
  requests:
    cpu: '200m'
    memory: '512Mi'

secrets:
  bigqueryCredentials: '/path/to/credentials.json'
  postgresCredentials:
    user: 'myuser'
    password: 'mypassword'

# File: de-onboarding/helm/transaction-pipeline/templates/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ .Chart.Name }}
  namespace: data-pipeline
spec:
  replicas: {{ .Values.replicaCount }}
  selector:
    matchLabels:
      app: {{ .Chart.Name }}
  template:
    metadata:
      labels:
        app: {{ .Chart.Name }}
    spec:
      containers:
      - name: fastapi
        image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
        ports:
        - containerPort: 8000
        resources:
          limits:
            cpu: {{ .Values.resources.limits.cpu }}
            memory: {{ .Values.resources.limits.memory }}
          requests:
            cpu: {{ .Values.resources.requests.cpu }}
            memory: {{ .Values.resources.requests.memory }}
        env:
        - name: GOOGLE_APPLICATION_CREDENTIALS
          value: {{ .Values.secrets.bigqueryCredentials }}
        - name: POSTGRES_USER
          value: {{ .Values.secrets.postgresCredentials.user }}
        - name: POSTGRES_PASSWORD
          value: {{ .Values.secrets.postgresCredentials.password }}

# File: de-onboarding/helm/transaction-pipeline/templates/service.yaml
apiVersion: v1
kind: Service
metadata:
  name: { { .Chart.Name } }
  namespace: data-pipeline
spec:
  selector:
    app: { { .Chart.Name } }
  ports:
    - protocol: TCP
      port: { { .Values.service.port } }
      targetPort: 8000
  type: { { .Values.service.type } }

# File: de-onboarding/helm/transaction-pipeline/templates/secrets.yaml
apiVersion: v1
kind: Secret
metadata:
  name: pipeline-secrets
  namespace: data-pipeline
type: Opaque
data:
  bigqueryCredentials: { { .Values.secrets.bigqueryCredentials | b64enc } }
  postgresUser: { { .Values.secrets.postgresCredentials.user | b64enc } }
  postgresPassword:
    { { .Values.secrets.postgresCredentials.password | b64enc } }

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: summary_sales
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:
    - fact_sales
    - pg_sales
    - summary_sales
orchestration:
  airflow_dag: transform_sales_mart
  schedule: '0 0 * * *'
security:
  pii_fields:
    - transaction_id
  jwt_secret: 'your-secret-key'
kubernetes:
  namespace: data-pipeline
  helm_chart: transaction-pipeline
  resources:
    cpu: '1'
    memory: '2Gi'

# File: de-onboarding/app/main.py (as above)

# File: de-onboarding/tests/test_pipeline.py
from typing import List
import pytest
from fastapi.testclient import TestClient
from app.main import app, TransactionSummary, QueryParams
import yaml
from google.cloud import bigquery
from unittest.mock import patch
import os
import logging
import hashlib
import psycopg2

client = TestClient(app)

@pytest.fixture
def config():
    """Load pipeline_config.yaml."""
    with open("data/pipeline_config.yaml", "r") as f:
        return yaml.safe_load(f)

@pytest.fixture
def auth_token():
    """Get JWT token for authenticated requests."""
    response = client.post("/token", data={"username": "admin", "password": "password123"})
    return response.json()["access_token"]

@pytest.fixture
def pg_conn():
    """Connect to PostgreSQL."""
    config = yaml.safe_load(open("data/pipeline_config.yaml"))
    pg_config = config["data_warehouse"]["postgres"]
    secret_data = utils.read_k8s_secret(pg_config["secretName"], "data-pipeline")
    conn = psycopg2.connect(
        dbname=pg_config["dbname"],
        host=pg_config["host"],
        port=pg_config["port"],
        user=secret_data["user"],
        password=secret_data["password"]
    )
    yield conn
    conn.close()

def test_login():
    """Test successful login."""
    response = client.post("/token", data={"username": "admin", "password": "password123"})
    assert response.status_code == 200
    assert "access_token" in response.json()

def test_unauthorized_access():
    """Test unauthorized access to trends endpoint."""
    response = client.get("/sales/trends")
    assert response.status_code == 401

def test_sales_trends(auth_token: str, config: dict):
    """Test BigQuery sales trends endpoint."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    response = client.get("/sales/trends?product_prefix=Halal&sale_month=2023-10-01", headers=headers)
    assert response.status_code == 200
    summaries: List[TransactionSummary] = [TransactionSummary(**item) for item in response.json()]
    assert len(summaries) > 0
    for summary in summaries:
        assert summary.product.startswith(config["ingestion"]["validation"]["product_prefix"])
        assert summary.total_amount > 0
        assert summary.transaction_count > 0

def test_pg_sales_trends(auth_token: str, config: dict, pg_conn):
    """Test PostgreSQL sales trends endpoint."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    response = client.get("/sales/pg-trends?product_prefix=Halal&sale_month=2023-10-01", headers=headers)
    assert response.status_code == 200
    summaries: List[TransactionSummary] = [TransactionSummary(**item) for item in response.json()]
    assert len(summaries) > 0
    for summary in summaries:
        assert summary.product.startswith(config["ingestion"]["validation"]["product_prefix"])
        assert summary.total_amount > 0
        assert summary.transaction_count > 0

def test_invalid_query_params(auth_token: str):
    """Test invalid query parameters."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    response = client.get("/sales/trends?product_prefix=Invalid@&sale_month=invalid", headers=headers)
    assert response.status_code == 422  # Validation error

def test_dashboard(auth_token: str):
    """Test dashboard rendering with PII masking."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    response = client.get("/", headers=headers)
    assert response.status_code == 200
    assert "Sales Trends Dashboard" in response.text
    assert any(hashlib.sha256(s.product.encode()).hexdigest()[:8] in response.text for s in [TransactionSummary(transaction_id="T001", product="Halal Laptop", sale_month="2023-10-01", total_amount=1999.98, avg_price=999.99, transaction_count=1)])

def test_bigquery_error(auth_token: str):
    """Test handling of BigQuery schema errors."""
    with patch("google.cloud.bigquery.Client.query", side_effect=Exception("Invalid schema")):
        headers = {"Authorization": f"Bearer {auth_token}"}
        response = client.get("/sales/trends", headers=headers)
        assert response.status_code == 500
        assert "Query error: Invalid schema" in response.text

def test_postgres_error(auth_token: str, pg_conn):
    """Test handling of PostgreSQL connection errors."""
    with patch("psycopg2.connect", side_effect=psycopg2.Error("Connection refused")):
        headers = {"Authorization": f"Bearer {auth_token}"}
        response = client.get("/sales/pg-trends", headers=headers)
        assert response.status_code == 500
        assert "Query error: Connection refused" in response.text

def test_response_time_performance(auth_token: str, caplog):
    """Test API response time is logged and under 1 second."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    with caplog.at_level(logging.INFO):
        response = client.get("/sales/trends", headers=headers)
        assert response.status_code == 200
        for record in caplog.records:
            if "API response time" in record.message:
                duration = float(record.message.split(": ")[1].split(" ")[0])
                assert duration < 1.0, f"Response time {duration}s exceeds 1s"

def test_query_performance(auth_token: str):
    """Test BigQuery query performance (bytes processed < 10MB)."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    with patch("google.cloud.bigquery.Client.query") as mock_query:
        mock_job = mock_query.return_value
        mock_job.total_bytes_processed = 5 * 1024 * 1024  # 5MB
        mock_job.result.return_value = [
            {"transaction_id": "hash", "product": "Halal Laptop", "sale_month": "2023-10-01", "total_amount": 1999.98, "avg_price": 999.99, "transaction_count": 1}
        ]
        response = client.get("/sales/trends", headers=headers)
        assert response.status_code == 200
        assert mock_job.total_bytes_processed < 10 * 1024 * 1024, "Query exceeds 10MB"

# File: de-onboarding/Dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY app/ .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# File: de-onboarding/requirements.txt
fastapi==0.95.0
uvicorn==0.20.0
google-cloud-bigquery==3.4.0
apache-airflow==2.5.0
pytest==7.2.0
pyyaml==6.0
pydantic==1.10.0
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0
jinja2==3.1.2
psycopg2-binary==2.9.5
dbt-core==1.4.0
dbt-postgres==1.4.0
pandas-gbq==0.17.9
kubernetes==25.3.0

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

[
  {
    "transaction_id": "<64-char-hash>",
    "product": "Halal Laptop",
    "sale_month": "2023-10-01",
    "total_amount": 1999.98,
    "avg_price": 999.99,
    "transaction_count": 1
  }
]

[
  {
    "transaction_id": "<hash>",
    "product": "Halal Laptop",
    "sale_month": "2023-10-01",
    "total_amount": 1999.98,
    "avg_price": 999.99,
    "transaction_count": 1
  }
]

# File: de-onboarding/app/main.py (partial)
class QueryParams(BaseModel):
    product_prefix: str = Field(default="Halal", pattern=r"^[a-zA-Z0-9]+$")
    min_amount: float = Field(default=0.0, ge=0.0)
    max_quantity: int = Field(default=1000, ge=0)
    sale_month: str = Field(default="2023-10-01", pattern=r"^\d{4}-\d{2}-\d{2}$")
    min_avg_price: float = Field(default=0.0, ge=0.0)
    max_avg_price: float = Field(default=10000.0, ge=0.0)

@app.get("/sales/trends", response_model=List[TransactionSummary])
async def get_sales_trends(
    params: QueryParams = Depends(),
    current_user: Dict[str, Any] = Depends(get_current_user)
) -> List[TransactionSummary]:
    """Get sales trends from BigQuery summary_sales view with avg_price filter."""
    start_time = time.time()
    logger.info(f"User {current_user['username']} queried trends: product_prefix={params.product_prefix}, min_amount={params.min_amount}, max_quantity={params.max_quantity}, sale_month={params.sale_month}, min_avg_price={params.min_avg_price}, max_avg_price={params.max_avg_price}")
    try:
        query = """
        SELECT
            transaction_id,
            product,
            sale_month,
            total_amount,
            avg_price,
            transaction_count
        FROM `hijra-project.transactions_mart.summary_sales`
        WHERE product LIKE @prefix || '%'
        AND sale_month = @sale_month
        AND avg_price BETWEEN @min_avg_price AND @max_avg_price
        """
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                {"name": "prefix", "parameterType": {"type": "STRING"}, "parameterValue": {"value": params.product_prefix}},
                {"name": "sale_month", "parameterType": {"type": "STRING"}, "parameterValue": {"value": params.sale_month}},
                {"name": "min_avg_price", "parameterType": {"type": "FLOAT"}, "parameterValue": {"value": params.min_avg_price}},
                {"name": "max_avg_price", "parameterType": {"type": "FLOAT"}, "parameterValue": {"value": params.max_avg_price}}
            ]
        )
        logger.info(f"Executing BigQuery summary query")
        query_job = bq_client.query(query, job_config=job_config)
        results = [
            TransactionSummary(
                transaction_id=row["transaction_id"],
                product=row["product"],
                sale_month=row["sale_month"],
                total_amount=row["total_amount"],
                avg_price=row["avg_price"],
                transaction_count=row["transaction_count"]
            ) for row in query_job
        ]
        logger.info(f"Retrieved {len(results)} sales summaries")
        return results
    except Exception as e:
        logger.error(f"BigQuery query failed: {str(e)}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Query error: {str(e)}")
    finally:
        duration = time.time() - start_time
        logger.info(f"API response time: {duration:.2f} seconds")

# File: de-onboarding/tests/test_pipeline.py (partial)
def test_sales_trends_avg_price(auth_token: str, config: dict):
    """Test sales trends with avg_price filter."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    response = client.get("/sales/trends?product_prefix=Halal&sale_month=2023-10-01&min_avg_price=500&max_avg_price=1000", headers=headers)
    assert response.status_code == 200
    summaries: List[TransactionSummary] = [TransactionSummary(**item) for item in response.json()]
    assert len(summaries) > 0
    assert all(500 <= summary.avg_price <= 1000 for summary in summaries)

# File: de-onboarding/app/main.py (partial)
@app.get("/sales/pg-trends", response_model=List[TransactionSummary])
async def get_pg_sales_trends(
    params: QueryParams = Depends(),
    current_user: Dict[str, Any] = Depends(get_current_user)
) -> List[TransactionSummary]:
    """Get sales trends from PostgreSQL pg_sales table with result count logging."""
    start_time = time.time()
    logger.info(f"User {current_user['username']} queried PostgreSQL trends: product_prefix={params.product_prefix}, min_amount={params.min_amount}, max_quantity={params.max_quantity}, sale_month={params.sale_month}")
    try:
        conn = psycopg2.connect(**pg_conn_params)
        cursor = conn.cursor()
        query = """
        SELECT
            transaction_id,
            product,
            TO_CHAR(sale_date, 'YYYY-MM-DD') AS sale_month,
            SUM(amount) AS total_amount,
            AVG(price) AS avg_price,
            COUNT(DISTINCT transaction_id) AS transaction_count
        FROM pg_sales
        WHERE product LIKE %s || '%%'
        AND TO_CHAR(sale_date, 'YYYY-MM-DD') = %s
        GROUP BY transaction_id, product, TO_CHAR(sale_date, 'YYYY-MM-DD')
        HAVING SUM(amount) >= %s AND SUM(quantity) <= %s
        """
        cursor.execute(query, (params.product_prefix, params.sale_month, params.min_amount, params.max_quantity))
        results = [
            TransactionSummary(
                transaction_id=row[0],
                product=row[1],
                sale_month=row[2],
                total_amount=row[3],
                avg_price=row[4],
                transaction_count=row[5]
            ) for row in cursor.fetchall()
        ]
        cursor.close()
        conn.close()
        logger.info(f"User {current_user['username']} retrieved {len(results)} PostgreSQL sales summaries")
        return results
    except Exception as e:
        logger.error(f"PostgreSQL query failed: {str(e)}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Query error: {str(e)}")
    finally:
        duration = time.time() - start_time
        logger.info(f"API response time: {duration:.2f} seconds")

# File: de-onboarding/tests/test_pipeline.py (partial)
def test_pg_result_count_logging(auth_token: str, caplog, pg_conn):
    """Test result count logging in PostgreSQL trends endpoint."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    with caplog.at_level(logging.INFO):
        response = client.get("/sales/pg-trends", headers=headers)
        assert response.status_code == 200
        assert any("retrieved" in record.message and "PostgreSQL sales summaries" in record.message for record in caplog.records)

# File: de-onboarding/helm/transaction-pipeline/values.yaml (partial)
replicaCount: 3
resources:
  limits:
    cpu: '500m'
    memory: '1Gi'
  requests:
    cpu: '200m'
    memory: '512Mi'

helm install pipeline helm/transaction-pipeline --namespace data-pipeline
kubectl describe pods --namespace data-pipeline | grep -E 'Limits|Requests'

# File: de-onboarding/tests/test_pipeline.py (partial)
def test_query_performance(auth_token: str):
    """Test BigQuery query performance (bytes processed < 10MB)."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    with patch("google.cloud.bigquery.Client.query") as mock_query:
        mock_job = mock_query.return_value
        mock_job.total_bytes_processed = 5 * 1024 * 1024  # 5MB
        mock_job.result.return_value = [
            {"transaction_id": "hash", "product": "Halal Laptop", "sale_month": "2023-10-01", "total_amount": 1999.98, "avg_price": 999.99, "transaction_count": 1}
        ]
        response = client.get("/sales/trends", headers=headers)
        assert response.status_code == 200
        assert mock_job.total_bytes_processed < 10 * 1024 * 1024, "Query exceeds 10MB"
        print(f"Query processed: {mock_job.total_bytes_processed / 1024 / 1024}MB")

# File: de-onboarding/tests/test_pipeline.py (partial)
def test_pg_response_time_performance(auth_token: str, caplog, pg_conn):
    """Test PostgreSQL API response time is logged and under 1 second."""
    headers = {"Authorization": f"Bearer {auth_token}"}
    with caplog.at_level(logging.INFO):
        response = client.get("/sales/pg-trends", headers=headers)
        assert response.status_code == 200
        for record in caplog.records:
            if "API response time" in record.message:
                duration = float(record.message.split(": ")[1].split(" ")[0])
                assert duration < 1.0, f"Response time {duration}s exceeds 1s"