How to Implement Incremental Load with Watermark by updated_at in Python

import sqlite3
from datetime import datetime, timedelta


def watermark_incremental_load(db_path, table_name, last_watermark, source_data):
    """Load only rows with updated_at greater than the last watermark."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Create table if it doesn't exist
    cursor.execute(f"""
        CREATE TABLE IF NOT EXISTS {table_name} (
            id INTEGER PRIMARY KEY,
            name TEXT,
            updated_at TEXT
        )
    """)

    new_watermark = last_watermark
    loaded_count = 0

    for row in source_data:
        if row["updated_at"] > last_watermark:
            cursor.execute(
                f"INSERT INTO {table_name} (id, name, updated_at) VALUES (?, ?, ?)",
                (row["id"], row["name"], row["updated_at"])
            )
            loaded_count += 1
            # Update watermark to the max timestamp seen
            if row["updated_at"] > new_watermark:
                new_watermark = row["updated_at"]

    conn.commit()
    conn.close()

    return {
        "loaded_count": loaded_count,
        "new_watermark": new_watermark,
        "skipped_count": len(source_data) - loaded_count
    }


if __name__ == "__main__":
    # Test data: some rows are new, some are already loaded
    last_run = "2024-01-15 10:00:00"
    source = [
        {"id": 1, "name": "Alice", "updated_at": "2024-01-15 08:00:00"},
        {"id": 2, "name": "Bob", "updated_at": "2024-01-15 09:30:00"},
        {"id": 3, "name": "Charlie", "updated_at": "2024-01-15 10:30:00"},
        {"id": 4, "name": "Diana", "updated_at": "2024-01-15 11:00:00"}
    ]

    result = watermark_incremental_load(
        "example.db", "users", last_run, source
    )
    print(result)