import pandas as pd
import io
 
# ---------------------------------------------------------------
# Non-idempotent pipeline: append-only load doubles data on re-run
# ---------------------------------------------------------------
 
def non_idempotent_store() -> dict:
    """Simulates an append-only destination."""
    return {"rows": []}
 
def load_append(df: pd.DataFrame, store: dict) -> int:
    """Appends rows every time — NOT idempotent."""
    new_rows = df.to_dict("records")
    store["rows"].extend(new_rows)
    return len(store["rows"])
 
 
SOURCE = pd.DataFrame({"id": [1, 2, 3], "amount": [100, 200, 300]})
 
store = non_idempotent_store()
run1 = load_append(SOURCE, store)
print(f"After run 1: {run1} rows")  # 3
 
run2 = load_append(SOURCE, store)
print(f"After run 2: {run2} rows")  # 6 — DUPLICATED!
 
 
# ---------------------------------------------------------------
# Idempotent pipeline: overwrite replaces destination each run
# ---------------------------------------------------------------
 
class OverwriteStore:
    """Destination that replaces content on each write."""
    def __init__(self):
        self.df = pd.DataFrame()
 
    def write(self, df: pd.DataFrame) -> None:
        self.df = df.copy()
 
    def read(self) -> pd.DataFrame:
        return self.df
 
 
def load_overwrite(df: pd.DataFrame, store: OverwriteStore) -> int:
    """Overwrites destination — idempotent."""
    store.write(df)
    return len(store.df)
 
 
overwrite_store = OverwriteStore()
run1 = load_overwrite(SOURCE, overwrite_store)
print(f"\nOverwrite run 1: {run1} rows")  # 3
 
run2 = load_overwrite(SOURCE, overwrite_store)
print(f"Overwrite run 2: {run2} rows")   # 3 — same result, no duplicates!

import pandas as pd
import io
 
# Simulated source with timestamps
SOURCE_DATA = """id,amount,product,created_at
1,100,apple,2024-01-01
2,200,orange,2024-01-02
3,300,grape,2024-01-03
4,400,melon,2024-01-04
5,500,cherry,2024-01-05
6,600,plum,2024-01-06
7,700,mango,2024-01-07
"""
 
 
def extract_incremental(raw_data: str, watermark: str) -> pd.DataFrame:
    """
    Extract only records with created_at > watermark.
    watermark: ISO date string (e.g. '2024-01-03')
    """
    df = pd.read_csv(io.StringIO(raw_data))
    df["created_at"] = pd.to_datetime(df["created_at"])
    cutoff = pd.to_datetime(watermark)
    return df[df["created_at"] > cutoff].reset_index(drop=True)
 
 
def run_incremental_pipeline(raw_data: str, state: dict) -> dict:
    """
    Run incremental extract. Updates state with new watermark.
    state: {"last_processed_at": "2024-01-01"}
    """
    watermark = state.get("last_processed_at", "1900-01-01")
    print(f"\nRun starting. Watermark: {watermark}")
 
    batch = extract_incremental(raw_data, watermark)
 
    if len(batch) == 0:
        print("  No new records found.")
        return state
 
    # Simulate load
    print(f"  Processing {len(batch)} new records: {list(batch['product'])}")
 
    # Update watermark to the max created_at in this batch
    new_watermark = batch["created_at"].max().strftime("%Y-%m-%d")
    state["last_processed_at"] = new_watermark
    print(f"  New watermark: {new_watermark}")
 
    return state
 
 
# Simulate three pipeline runs
pipeline_state = {"last_processed_at": "2024-01-02"}
 
pipeline_state = run_incremental_pipeline(SOURCE_DATA, pipeline_state)
# Processes records 3, 4, 5, 6, 7
 
pipeline_state = run_incremental_pipeline(SOURCE_DATA, pipeline_state)
# Processes 0 records — watermark is now 2024-01-07, no newer data
 
# Simulate new records arriving
NEW_RECORDS = SOURCE_DATA + "8,800,kiwi,2024-01-08\n"
pipeline_state = run_incremental_pipeline(NEW_RECORDS, pipeline_state)
# Processes only record 8

import pandas as pd
import sqlite3
 
# In-memory database
conn = sqlite3.connect(":memory:")
 
# Initial data in the database
initial_products = pd.DataFrame({
    "id":    [1, 2, 3],
    "name":  ["Apple",  "Orange", "Grape"],
    "price": [1.00,     1.50,     2.00],
})
initial_products.to_sql("products", conn, if_exists="replace", index=False)
 
print("Initial state:")
print(pd.read_sql("SELECT * FROM products", conn))
 
# New batch: updates product 2 (price change), adds product 4 (new)
new_batch = pd.DataFrame({
    "id":    [2, 4],
    "name":  ["Orange", "Mango"],
    "price": [1.75,     3.00],   # id=2 price changed
})
 
 
# ---------------------------------------------------------------
# Upsert using pandas merge + combine_first
# ---------------------------------------------------------------
 
def upsert_pandas(new_df: pd.DataFrame, conn: sqlite3.Connection,
                  table: str, key_col: str) -> int:
    """
    Upsert new_df into existing table using pandas merge.
    New rows are inserted, existing rows are updated.
    """
    existing = pd.read_sql(f"SELECT * FROM {table}", conn)
 
    # Set key as index for combine_first
    merged = (
        new_df.set_index(key_col)
        .combine_first(existing.set_index(key_col))
        .reset_index()
    )
 
    # Overwrite the table with merged result
    merged.to_sql(table, conn, if_exists="replace", index=False)
    return len(merged)
 
 
total = upsert_pandas(new_batch, conn, "products", key_col="id")
print("\nAfter upsert:")
print(pd.read_sql("SELECT * FROM products ORDER BY id", conn))
print(f"Total rows: {total}")
 
 
# ---------------------------------------------------------------
# Upsert using SQLite INSERT OR REPLACE
# ---------------------------------------------------------------
 
conn2 = sqlite3.connect(":memory:")
conn2.execute("""
    CREATE TABLE products (
        id INTEGER PRIMARY KEY,
        name TEXT,
        price REAL
    )
""")
conn2.execute("INSERT INTO products VALUES (1, 'Apple', 1.00)")
conn2.execute("INSERT INTO products VALUES (2, 'Orange', 1.50)")
conn2.execute("INSERT INTO products VALUES (3, 'Grape', 2.00)")
conn2.commit()
 
# INSERT OR REPLACE treats primary key conflict as a replace operation
upsert_rows = [(2, "Orange", 1.75), (4, "Mango", 3.00)]
conn2.executemany(
    "INSERT OR REPLACE INTO products (id, name, price) VALUES (?, ?, ?)",
    upsert_rows
)
conn2.commit()
 
print("\nAfter INSERT OR REPLACE:")
result = conn2.execute("SELECT * FROM products ORDER BY id").fetchall()
for row in result:
    print(f"  id={row[0]}, name={row[1]}, price={row[2]}")
 
conn.close()
conn2.close()