pipeline:
  name: "customer_orders"
  schedule: "0 6 * * *"

source:
  format: "json"
  path: "/mnt/raw/customer_orders/"

destination:
  database: "analytics"
  table: "customer_orders"

from src.config_manager import ConfigManager
from src.bronze_loader import BronzeLoader
from src.silver_transformer import SilverTransformer
from src.gold_aggregator import GoldAggregator

config = ConfigManager("configs/pipeline_config.yaml")

# Bronze: ingest raw data
bronze = BronzeLoader(config)
bronze_df = bronze.run()

# Silver: clean and deduplicate
silver = SilverTransformer(config)
silver_df = silver.run()

# Gold: aggregate for analytics
gold = GoldAggregator(config)
gold_df = gold.run()

from src.quality_gate import QualityGate

gate = QualityGate(config, rules_path="configs/quality_rules.yaml")
result = gate.validate(silver_df, layer="silver")

if not result.passed:
    raise RuntimeError(f"Quality gate failed: {result.failures}")

from src.extractors.jdbc_extractor import JDBCExtractor

extractor = JDBCExtractor(spark, config={
    "jdbc": {
        "url": "jdbc:postgresql://db.example.com:5432/analytics",
        "table": "public.orders",
        "user": "${DB_USER}",
        "password": "${DB_PASSWORD}",
        "partition_column": "order_id",
        "num_partitions": 16,
    }
})
df = extractor.extract()

from src.extractors.file_extractor import FileExtractor

extractor = FileExtractor(spark, config={
    "file_source": {
        "format": "parquet",
        "path": "/mnt/landing/orders/",
        "incremental": True,
        "watermark_path": "/mnt/checkpoints/orders_wm.json",
    }
})
df = extractor.extract()

from src.extractors.api_extractor import APIExtractor

extractor = APIExtractor(spark, config={
    "api": {
        "base_url": "https://api.example.com/v2",
        "endpoint": "/customers",
        "auth": {"type": "bearer", "token": "${API_TOKEN}"},
        "pagination": {"strategy": "cursor", "page_size": 200},
        "rate_limit": {"requests_per_second": 5, "burst_size": 10},
    }
})
df = extractor.extract()

from src.transformers.scd_handler import SCDHandler

handler = SCDHandler(spark, config={
    "scd": {
        "target_table": "silver.customers",
        "primary_keys": ["customer_id"],
        "scd_type": 2,
        "tracked_columns": ["name", "email", "tier"],
        "change_detection": "hash",
        "soft_delete": True,
    }
})

result = handler.merge(incoming_df)
print(result.summary)
# → SCD2 merge into silver.customers: 150 inserted, 42 updated, 42 expired, 308 unchanged (3.21s)

from src.transformers.deduplication import DeduplicationEngine

engine = DeduplicationEngine(spark, config={
    "dedup": {
        "strategy": "window",
        "key_columns": ["order_id"],
        "order_column": "_ingested_at",
        "order_direction": "desc",
        "keep": "first",
    }
})

clean_df, rejected_df = engine.deduplicate(raw_df)

from src.utils.metrics_collector import MetricsCollector

collector = MetricsCollector(spark, config={
    "metrics": {"target_table": "ops.pipeline_metrics", "sla_minutes": 60}
}, pipeline_name="customer_orders", batch_id="batch_20250315")

collector.record_event("bronze", rows=50000, duration=12.5)
collector.record_event("silver", rows=49500, duration=45.2)
collector.record_quality_score("silver", passed=8, failed=0, total=8)

summary = collector.finalise()
# → Pipeline: customer_orders | Batch: batch_20250315
#   Duration: 66.4s (SLA: 60min — MET)
#   Rows: 99,500 processed, 500 rejected
#   Quality: 100.0%

from src.utils.lineage_tracker import LineageTracker

tracker = LineageTracker(pipeline_name="customer_orders")
tracker.add_source("raw_orders", source_type="file", path="/mnt/raw/")
tracker.add_transformation(
    name="silver_clean", inputs=["raw_orders"], outputs=["silver_orders"]
)
tracker.add_target("gold_metrics", target_type="delta_table")

# Export as Mermaid diagram
print(tracker.to_mermaid())
# → graph LR
#     raw_orders([raw_orders])
#     silver_clean{{silver_clean}}
#     gold_metrics[gold_metrics]
#     raw_orders -->|consumes| silver_clean
#     silver_clean -->|produces| gold_metrics

# Query upstream lineage
print(tracker.get_upstream("gold_metrics"))
# → ['silver_clean', 'raw_orders']