From four source databases to one governed BI layer

"""
Migration flow — Shipments: LegacyRoad (TMS1) -> NexHaul (TMS2) -> warehouse.

Extract dbo.Shipments from the on-prem SQL Server behind SSMS, conform it
to the TMS2 data model through a SnapLogic pipeline (field renames + type
casts live there, version-controlled with the pipeline), merge with native
TMS2 records, and load the unified table to the warehouse.
"""

import io
import os

import pandas as pd
import pyodbc
import requests
from prefect import flow, task, get_run_logger
from prefect.blocks.system import Secret

SOURCE_TABLE = "dbo.Shipments"                # TMS1, SQL Server (SSMS)
TMS2_TABLE = "tms2.shipments"                  # TMS2, already conformed
WAREHOUSE_TABLE = "dw.fact_shipment"        # load target
NATURAL_KEY = "shipment_number"          # dedupe key after conforming
SNAPLOGIC_TASK_URL = (
    "https://elastic.snaplogic.com/api/1/rest/slsched/feed/FreightIntel"
    "/migration/shipments_conform_task"
)

# Contract with the SnapLogic pipeline: exact columns + types it must emit.
EXPECTED_SCHEMA = {
    "shipment_id": "bigint",
    "shipment_number": "varchar(12)",
    "pickup_date": "date",
    "delivery_date": "date",
    "origin_city": "varchar(40)",
    "origin_region": "char(2)",
    "dest_city": "varchar(40)",
    "dest_region": "char(2)",
    "client_code": "varchar(8)",
    "equipment_type": "varchar(10)",
    "cross_border": "boolean",
    "loaded_miles": "int",
    "total_revenue": "decimal(12,2)",
    "status": "varchar(12)",
}


def sqlserver_conn(dsn_secret: str) -> pyodbc.Connection:
    """Connection via a DSN stored in a Prefect Secret block (no creds in code)."""
    return pyodbc.connect(Secret.load(dsn_secret).get(), timeout=30)


@task(retries=3, retry_delay_seconds=60)
def extract_tms1() -> pd.DataFrame:
    """Pull the legacy table exactly as stored — no transforms at extract time."""
    logger = get_run_logger()
    with sqlserver_conn("tms1-ssms-dsn") as conn:
        df = pd.read_sql(f"SELECT * FROM {SOURCE_TABLE};", conn)
    logger.info("TMS1 extract: %s rows from %s", len(df), SOURCE_TABLE)
    return df


@task(retries=3, retry_delay_seconds=120)
def conform_via_snaplogic(tms1_df: pd.DataFrame) -> pd.DataFrame:
    """
    POST the raw TMS1 rows to the SnapLogic triggered task that owns the
    field-name and data-type mapping, and read back TMS2-shaped records.
    """
    logger = get_run_logger()
    response = requests.post(
        SNAPLOGIC_TASK_URL,
        params={"bearer_token": Secret.load("snaplogic-token").get()},
        json=tms1_df.to_dict(orient="records"),
        timeout=600,
    )
    response.raise_for_status()
    conformed = pd.DataFrame(response.json())

    missing = set(EXPECTED_SCHEMA) - set(conformed.columns)
    if missing:
        raise ValueError(f"SnapLogic output missing columns: {sorted(missing)}")
    logger.info("SnapLogic conformed %s rows to the TMS2 model", len(conformed))
    return conformed


@task(retries=3, retry_delay_seconds=60)
def extract_tms2() -> pd.DataFrame:
    """Native TMS2 records — already in the target model."""
    logger = get_run_logger()
    with sqlserver_conn("tms2-replica-dsn") as conn:
        df = pd.read_sql(f"SELECT * FROM {TMS2_TABLE};", conn)
    logger.info("TMS2 extract: %s rows from %s", len(df), TMS2_TABLE)
    return df


@task
def merge_systems(conformed_tms1: pd.DataFrame, tms2: pd.DataFrame) -> pd.DataFrame:
    """
    Union both systems on the shared model, then dedupe on the natural key.
    TMS2 wins collisions: it is the system of record for migrated entities.
    """
    logger = get_run_logger()
    merged = pd.concat(
        [conformed_tms1.assign(source_system="TMS1"), tms2.assign(source_system="TMS2")],
        ignore_index=True,
    )
    before = len(merged)
    merged = merged.sort_values("source_system").drop_duplicates(NATURAL_KEY, keep="last")
    logger.info("Merged %s rows -> %s after dedupe on %s", before, len(merged), NATURAL_KEY)
    return merged


@task(retries=2, retry_delay_seconds=300)
def load_warehouse(df: pd.DataFrame) -> int:
    """
    Bulk-load to a staging table, then swap into the target inside one
    transaction so downstream reports never see a half-loaded table.
    """
    logger = get_run_logger()
    staging = f"{WAREHOUSE_TABLE}_stg"
    with sqlserver_conn("warehouse-dsn") as conn:
        cursor = conn.cursor()
        cursor.fast_executemany = True
        cursor.execute(f"TRUNCATE TABLE {staging};")
        placeholders = ", ".join("?" for _ in df.columns)
        cursor.executemany(
            f"INSERT INTO {staging} ({', '.join(df.columns)}) VALUES ({placeholders})",
            df.itertuples(index=False, name=None),
        )
        cursor.execute(
            f"BEGIN TRAN; TRUNCATE TABLE {WAREHOUSE_TABLE}; "
            f"INSERT INTO {WAREHOUSE_TABLE} SELECT * FROM {staging}; COMMIT;"
        )
        conn.commit()
    logger.info("Loaded %s rows into %s", len(df), WAREHOUSE_TABLE)
    return len(df)


@flow(name="migrate-shipments", log_prints=True)
def migrate_shipments() -> None:
    tms1_raw = extract_tms1()
    conformed = conform_via_snaplogic(tms1_raw)
    tms2_native = extract_tms2()
    unified = merge_systems(conformed, tms2_native)
    load_warehouse(unified)


if __name__ == "__main__":
    # Local run; in production this deploys on a schedule via prefect.yaml.
    migrate_shipments()