Real-time E-commerce Inventory Sync Blueprint

import requests
import json
from datetime import datetime, timedelta
import psycopg2

# --- Configuration ---
ECOMMERCE_API_URL = "https://api.example-ecommerce.com/v1/inventory"
API_KEY = "YOUR_API_KEY"

DB_HOST = "localhost"
DB_NAME = "inventory_db"
DB_USER = "user"
DB_PASSWORD = "password"

# --- Database Connection ---
def get_db_connection():
    try:
        conn = psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
        return conn
    except psycopg2.Error as e:
        print(f"Error connecting to PostgreSQL: {e}")
        return None

# --- API Interaction ---
def fetch_inventory_data():
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    params = {
        "limit": 1000, # Example pagination
        "updated_since": (datetime.utcnow() - timedelta(minutes=15)).isoformat() + "Z" # Fetch recent updates
    }
    
    all_inventory = []
    page = 1
    while True:
        params['page'] = page
        try:
            response = requests.get(ECOMMERCE_API_URL, headers=headers, params=params)
            response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
            data = response.json()
            
            if not data.get('products'):
                break
                
            for item in data['products']:
                for variant in item.get('variants', []):
                    all_inventory.append({
                        'sku': variant.get('sku'),
                        'product_id': item.get('id'),
                        'variant_id': variant.get('id'),
                        'quantity': variant.get('inventory_quantity'),
                        'location_id': variant.get('inventory_item_id'), # Simplified mapping
                        'last_updated': datetime.utcnow().isoformat()
                    })
            
            # Basic pagination check - adjust based on API response structure
            if len(data.get('products', [])) < 1000:
                break
            page += 1
            
        except requests.exceptions.RequestException as e:
            print(f"Error fetching data from API: {e}")
            break
        except json.JSONDecodeError:
            print("Error decoding JSON response from API.")
            break
            
    return all_inventory

# --- Database Loading ---
def load_inventory_to_db(inventory_data):
    conn = get_db_connection()
    if not conn:
        return

    cur = conn.cursor()
    # Assuming a table named 'stg_inventory' exists with columns: sku, product_id, variant_id, quantity, location_id, last_updated
    sql = """
    INSERT INTO stg_inventory (sku, product_id, variant_id, quantity, location_id, last_updated)
    VALUES (%s, %s, %s, %s, %s, %s)
    ON CONFLICT (sku) DO UPDATE SET
        quantity = EXCLUDED.quantity,
        last_updated = EXCLUDED.last_updated
    """
    
    try:
        for item in inventory_data:
            cur.execute(sql, (
                item.get('sku'),
                item.get('product_id'),
                item.get('variant_id'),
                item.get('quantity'),
                item.get('location_id'),
                item.get('last_updated')
            ))
        conn.commit()
        print(f"Successfully loaded {len(inventory_data)} inventory records.")
    except psycopg2.Error as e:
        conn.rollback()
        print(f"Error loading data to database: {e}")
    finally:
        cur.close()
        conn.close()

# --- Main Execution ---
def main():
    print("Starting inventory sync...")
    inventory_data = fetch_inventory_data()
    if inventory_data:
        load_inventory_to_db(inventory_data)
    else:
        print("No inventory data fetched or an error occurred.")
    print("Inventory sync finished.")

if __name__ == "__main__":
    main()