Batch Upsert Thousands of Rows from Alteryx to a SharePoint List using Python

################################# # Configuration ################################# from ayx import Alteryx import pandas as pd import requests import time from datetime import datetime CONFIG = { "tenant_id": "your_tenant_id", "client_id": "your_client_id", "client_secret": "your_client_id", "site_id": "your_tenant.sharepoint.com,your_site_collection_id,your_list_id", "list_id": "your_list_id", "unique_field": "Title", # Match the list's title field "batch_size": 20, # Items per batch } # Logging Config def log_info(message): timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') log_entry = f"[OK] [{timestamp}] | INFO | PYTHON DELETE | {message}" print(log_entry) def log_error(message): timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') log_entry = f"[X] [{timestamp}] | ERROR | PYTHON DELETE | {message}" print(log_entry)

################################# # Definitions ################################# def authenticate(config): """Get access token for Graph API.""" token_url = f"https://login.microsoftonline.com/{config['tenant_id']}/oauth2/v2.0/token" response = requests.post( token_url, data={ "client_id": config["client_id"], "client_secret": config["client_secret"], "scope": "https://graph.microsoft.com/.default", "grant_type": "client_credentials", }, timeout=30, ) if response.status_code == 200: token = response.json().get("access_token") if token: return token raise Exception(f"Authentication failed: {response.text}") def make_request(method, url, headers, retries=3, **kwargs): """Make HTTP request""" for attempt in range(retries): try: response = requests.request(method, url, headers=headers, **kwargs) if response.status_code in [200, 201, 204]: return response # Handle throttling if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 10)) log_info(f"Throttled. Waiting {retry_after}s...") time.sleep(retry_after) continue # Handle service errors if response.status_code in [503, 504]: wait_time = min(2 ** attempt, 30) log_error(f"Service error. Retry in {wait_time}s...") time.sleep(wait_time) continue response.raise_for_status() except Exception as e: if attempt == retries - 1: raise time.sleep(2 ** attempt) raise Exception("Max retries exceeded") def get_existing_items(base_url, site_id, list_id, headers, unique_field): """Retrieve all existing items from SharePoint List.""" items = {} url = f"{base_url}/sites/{site_id}/lists/{list_id}/items?$expand=fields&$top=999" log_info("Fetching existing SharePoint items...") page_count = 0 while url: page_count += 1 response = make_request("GET", url, headers) data = response.json() # Process items for item in data.get("value", []): if unique_field in item.get("fields", {}): title = str(item["fields"][unique_field]) items[title] = item["id"] # Get next page URL url = data.get("@odata.nextLink") if url: time.sleep(0.2) # Small delay between pages log_info(f"Found {len(items)} items across {page_count} pages") return items def batch_delete_items(base_url, site_id, list_id, headers, item_ids, batch_size=20): """Delete items from SharePoint in batches using Graph API batch endpoint.""" if not item_ids: return pd.DataFrame() results = [] total_batches = (len(item_ids) + batch_size - 1) // batch_size log_info(f"Deleting {len(item_ids)} items in {total_batches} batches...") for batch_num in range(total_batches): start_idx = batch_num * batch_size end_idx = min(start_idx + batch_size, len(item_ids)) batch_ids = item_ids[start_idx:end_idx] # Build batch request batch_requests = [ { "id": str(i), "method": "DELETE", "url": f"/sites/{site_id}/lists/{list_id}/items/{item_id}", "headers": {"If-Match": "*"} } for i, item_id in enumerate(batch_ids) ] # Send batch request try: response = make_request( "POST", f"{base_url}/$batch", headers, json={"requests": batch_requests} ) batch_response = response.json() # Process responses for i, resp in enumerate(batch_response.get("responses", [])): status_code = resp.get("status", 0) results.append({ "item_id": batch_ids[i] if i < len(batch_ids) else "unknown", "status": "success" if status_code == 204 else "failed", "status_code": status_code, "batch": batch_num + 1 }) except Exception as e: log_error(f"Deletion Batch {batch_num + 1} error: {str(e)}") for item_id in batch_ids: results.append({ "item_id": item_id, "status": "failed", "status_code": 0, "error": str(e), "batch": batch_num + 1 }) # Delay in between batches if batch_num < total_batches - 1: time.sleep(0.5) return pd.DataFrame(results)

################################# # Execution ################################# try: log_info("SharePoint List Cleanup - Delete Items Not in Today's Data") # Read input data from Alteryx input_data = Alteryx.read("#1") unique_field = CONFIG["unique_field"] # Validate unique field exists if unique_field not in input_data.columns: raise Exception(f"Column '{unique_field}' not found in input data") # Get unique values from today's data todays_values = set(input_data[unique_field].astype(str).unique()) log_info(f"Today's data contains {len(todays_values)} unique rows") # Authenticate log_info("Authenticating...") token = authenticate(CONFIG) # Setup headers headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } base_url = "https://graph.microsoft.com/v1.0" # Get existing items from SharePoint existing_items = get_existing_items( base_url, CONFIG["site_id"], CONFIG["list_id"], headers, unique_field ) # Identify items to delete (exist in SharePoint but not in today's data) items_to_delete = [] for title, item_id in existing_items.items(): if title not in todays_values: items_to_delete.append(item_id) log_info(f"Items to delete: {len(items_to_delete)} of {len(existing_items)}") # Perform deletion if needed if items_to_delete: results = batch_delete_items( base_url, CONFIG["site_id"], CONFIG["list_id"], headers, items_to_delete, CONFIG["batch_size"] ) # Calculate statistics success_count = len(results[results["status"] == "success"]) failed_count = len(results[results["status"] == "failed"]) log_info(f"Deletion Complete - Success: {success_count}, Failed: {failed_count}") # Show failed items if any if failed_count > 0: failed = results[results["status"] == "failed"] log_error("Failed deletions (first 5):") for _, row in failed.head(5).iterrows(): log_error(f" Item ID {row['item_id']}: Status {row['status_code']}") # Output results overview Alteryx.write(results, 1) # Log output summary log_info(f"Items_Checked: {len(existing_items)}") log_info(f"Items_To_Delete: {len(items_to_delete)}") log_info(f"Delete_Success: {success_count}") log_info(f"Delete_Failed: {failed_count}") else: log_info("No items to delete - all SharePoint items are in today's data") # Log empty output summary log_info(f"Items_Checked: {len(existing_items)}") log_info("Items_To_Delete: 0") log_info("Delete_Success: 0") log_info("Delete_Failed: 0") except Exception as e: log_error(f"X ERROR: {str(e)}") error_df = pd.DataFrame([{ "Error": str(e), "Timestamp": datetime.now() }]) Alteryx.write(error_df, 1) raise

################################# # Configuration ################################# from ayx import Alteryx import pandas as pd import numpy as np import requests import time from datetime import datetime # get from environment variables in prod CONFIG = { "tenant_id": "your_tenant_id", "client_id": "your_client_id", "client_secret": "your_client_id", "site_id": "your_tenant.sharepoint.com,your_site_collection_id,your_list_id", "list_id": "your_list_id", "unique_field": "Title", # Match the list's title field } # SharePoint internal/system columns to exclude EXCLUDED_COLUMNS = [ 'Item Type', 'Path', 'Content Type', 'Modified', 'Created', 'Created By', 'Modified By', 'ID', 'Attachments', 'Edit', '_UIVersionString', 'File Type', 'App Modified By', 'App Created By', 'Compliance Asset Id', '_ComplianceFlags', '_ComplianceTag', '_ComplianceTagUserId', '_ComplianceTagWrittenTime', '_IsRecord', 'Item Child Count', 'Folder Child Count' ] # Logging Config def log_info(message): timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') log_entry = f"[OK] [{timestamp}] | INFO | PYTHON UPSERT | {message}" print(log_entry) def log_error(message): timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') log_entry = f"[X] [{timestamp}] | ERROR | PYTHON UPSERT | {message}" print(log_entry)

################################# # Definitions ################################# def authenticate(config): """Get access token using requests.""" token_url = f"https://login.microsoftonline.com/{config['tenant_id']}/oauth2/v2.0/token" response = requests.post( token_url, data={ "client_id": config["client_id"], "client_secret": config["client_secret"], "scope": "https://graph.microsoft.com/.default", "grant_type": "client_credentials" }, timeout=30 ) if response.status_code == 200: result = response.json() if "access_token" in result: return result["access_token"] raise Exception(f"Auth failed: {response.text}") class SharePointSync: """Production SharePoint sync for Alteryx.""" def __init__(self, token, site_id, list_id): self.headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Prefer": "return=representation" # Add this header for better responses } self.base = "https://graph.microsoft.com/v1.0" self.site_id = site_id self.list_id = list_id def request(self, method, url, retries=5, **kwargs): """HTTP request with retry logic.""" for i in range(retries): try: r = requests.request(method, url, headers=self.headers, **kwargs) if r.status_code in [200, 201, 204]: return r if r.status_code == 429: retry_after = int(r.headers.get('Retry-After', 10)) log_error(f"Throttled. Waiting {retry_after}s") time.sleep(retry_after) continue if r.status_code in [503, 504]: wait_time = min(2 ** i, 60) log_error(f"Service error {r.status_code}. Retry in {wait_time}s") time.sleep(wait_time) continue # Log detailed error for debugging if r.status_code == 400: log_error(f"Bad Request Error (400): {r.text}") elif r.status_code == 503: log_error(f"Service Unavailable (503): {r.text}") r.raise_for_status() except Exception as e: if i == retries - 1: raise time.sleep(min(2 ** i, 60)) raise Exception("Max retries exceeded") def get_items(self): """Get all existing items with pagination.""" items = [] url = f"{self.base}/sites/{self.site_id}/lists/{self.list_id}/items?$expand=fields&$top=999" while url: r = self.request("GET", url) data = r.json() items.extend(data.get("value", [])) url = data.get("@odata.nextLink") if url: time.sleep(0.3) return items def clean_fields_for_sharepoint(self, row_dict): """Clean and prepare fields for SharePoint API.""" cleaned = {} for key, value in row_dict.items(): # Skip excluded columns if key in EXCLUDED_COLUMNS: continue # Skip internal columns starting with underscore if key.startswith('_'): continue # Handle NaN and None values if pd.isna(value) or value is None: continue # Handle different data types if isinstance(value, bool): cleaned[key] = value elif isinstance(value, (int, float)): # Check for infinity using numpy if pd.notna(value) and not np.isinf(value): cleaned[key] = float(value) elif isinstance(value, pd.Timestamp): cleaned[key] = value.isoformat() else: # Convert to string and clean str_val = str(value).strip() if str_val and str_val.lower() not in ['nan', 'none', 'null']: cleaned[key] = str_val return cleaned def batch_upsert(self, source_df, unique_field): """Batch upsert DataFrame to SharePoint.""" log_info(f"Processing {len(source_df)} rows") log_info(f"Input columns ({len(source_df.columns)}): {list(source_df.columns)}") # Ensure unique field exists if unique_field not in source_df.columns: raise Exception(f"Unique field '{unique_field}' not found. Available: {list(source_df.columns)}") # Build lookup of existing items log_info("Fetching existing SharePoint items...") existing = self.get_items() lookup = {} for item in existing: if unique_field in item.get("fields", {}): key = str(item["fields"].get(unique_field)) lookup[key] = { "id": item["id"], "etag": item.get("eTag", item.get("@odata.etag", "*")) } log_info(f"Found {len(lookup)} existing items in SharePoint") results = [] batch_size = 20 total_batches = (len(source_df) + batch_size - 1) // batch_size for batch_num, i in enumerate(range(0, len(source_df), batch_size), 1): batch_df = source_df.iloc[i:i+batch_size] batch_requests = [] for idx, (_, row) in enumerate(batch_df.iterrows()): # Clean fields for SharePoint fields = self.clean_fields_for_sharepoint(row.to_dict()) # Get unique value for matching unique_val = str(fields.get(unique_field, "")) # Debug first item if i == 0 and idx == 0: log_info(f"First item: Title='{unique_val}'") log_info(f"Fields to send ({len(fields)}): {list(fields.keys())}") if unique_val in lookup: # Update existing item - use correct endpoint format item_info = lookup[unique_val] req = { "id": str(idx), "method": "PATCH", "url": f"/sites/{self.site_id}/lists/{self.list_id}/items/{item_info['id']}", "body": {"fields": fields}, "headers": { "Content-Type": "application/json", "If-Match": "*" # Force update regardless of eTag } } else: # Create new item req = { "id": str(idx), "method": "POST", "url": f"/sites/{self.site_id}/lists/{self.list_id}/items", "body": {"fields": fields}, "headers": {"Content-Type": "application/json"} } batch_requests.append(req) # Send batch request try: r = self.request("POST", f"{self.base}/$batch", json={"requests": batch_requests}) batch_response = r.json() # Process responses for resp in batch_response.get("responses", []): status_code = resp.get("status") error_msg = "" # Default to empty string instead of None if status_code not in [200, 201, 204]: # Parse error message error_body = resp.get("body", {}) if isinstance(error_body, dict): error = error_body.get("error", {}) error_msg = error.get("message", "Unknown error") else: error_msg = str(error_body) # Log first few errors in detail if len([r for r in results if r.get("status") == "failed"]) < 3: request_id = int(resp.get("id", 0)) if request_id < len(batch_requests): failed_req = batch_requests[request_id] method = failed_req['method'] log_error(f"\nError in {method} request (ID {request_id}):") log_error(f" Status: {status_code}") log_error(f" Message: {error_msg}") results.append({ "status": "success" if status_code in [200, 201, 204] else "failed", "id": resp.get("id"), "title": batch_df.iloc[int(resp.get("id"))][unique_field] if int(resp.get("id")) < len(batch_df) else "", "status_code": status_code, "error": error_msg if error_msg else "" }) except Exception as e: log_error(f"Batch error: {str(e)}") for req in batch_requests: results.append({ "status": "failed", "id": req["id"], "status_code": 0, "error": str(e) }) # Brief pause between batches if batch_num < total_batches: time.sleep(0.5) return pd.DataFrame(results)

################################# # Execution ################################# try: log_info("SharePoint List Sync (upsert) starting") # Read input from Alteryx input_data = Alteryx.read("#1") log_info(f"Loaded {len(input_data)} rows from Alteryx") # Remove any SharePoint-specific columns that might have been included columns_to_drop = [col for col in EXCLUDED_COLUMNS if col in input_data.columns] if columns_to_drop: input_data = input_data.drop(columns=columns_to_drop) # Authenticate log_info("Authenticating...") token = authenticate(CONFIG) log_info("Authentication successful") # Initialize sync sync = SharePointSync(token, CONFIG["site_id"], CONFIG["list_id"]) # Perform batch upsert log_info("Starting sync...") results = sync.batch_upsert(input_data, CONFIG["unique_field"]) # Replace any None/NaN values in results with empty strings results = results.fillna("") # Calculate statistics success_count = len(results[results["status"] == "success"]) failed_count = len(results[results["status"] == "failed"]) log_info(f"Sync Complete - Success: {success_count}, Failed: {failed_count}") # Show specific errors if any if failed_count > 0: failed_items = results[results["status"] == "failed"] log_error("Failed items (first 5):") for _, item in failed_items.head(5).iterrows(): if item['error']: log_error(f" ID {item['id']}: {item['error']}") # Output results to Alteryx Alteryx.write(results, 1) # Also output summary summary = pd.DataFrame([{ "Status": "Completed", "TotalRows": len(input_data), "Success": success_count, "Failed": failed_count, "Timestamp": datetime.now() }]) Alteryx.write(summary, 2) except Exception as e: log_error(f"ERROR: {str(e)}") error_df = pd.DataFrame([{ "Error": str(e), "Timestamp": datetime.now() }]) Alteryx.write(error_df, 1) raise