Adjusting CloudWorks Integration Schedules Using Python

AnaplanOEG

Intended Audience

Level of Difficulty: Intermediate

• Requires familiarity with APIs and python scripting

Resources Required:

• Internal Expertise: Python Developer / API Expert
• Tools Needed: Python, CloudWorks, Code Editor
• Access Requirements: CloudWorks, Anaplan Integration Admin

Estimated Level of Effort (LoE): Low

• A few hours to a day

Introduction

Customers who rely on Anaplan CloudWorks for scheduled integrations may need to shift their schedules forward or backward by 1 hour due to operational needs.

Scope

This guide provides a simple Python-based solution to:

• Retrieve all CloudWorks integrations with schedules
• Bulk adjust your schedules forward (+1 hour) or backward (-1 hour)
• Easily configure and execute the process with minimal technical effort

It does not delve into the basics of Python programming; familiarity with this languages is assumed.

Understanding CloudWorks Scheduling and DST Relevance

Integration and integration flow schedules auto-calculate the slot for the next run. This is based on your selected time zone, and the calculation factors daylight savings time (DST).

Solution Overview

The solution consists of:

1. get_cloudworks_integrations.py – Retrieves a list of integrations with schedules
2. config.json – Stores authentication details and integration IDs
3. adjust_cloudworks_schedules.py – Adjusts schedules by +1 or -1 hour

Step-by-Step Guide

Prerequisites

Before running the scripts, ensure you have:

• Python 3.8 or later installed
• API access to Anaplan CloudWorks
• CloudWorks integrations already scheduled
• Basic knowledge of running Python scripts

Step 1: Install Required Python Packages

pip install requests pandas pytz

Step 2: Retrieve Your CloudWorks Integrations

Use get_cloudworks_integrations.py to list all integrations that have schedules.

Example Script

import requests
import json
import os
import base64
import pandas as pd

def get_config():
    """ Load settings from config.json file, ensuring correct path resolution. """
    script_dir = os.path.dirname(os.path.realpath(__file__))  # Get the script's directory
    config_path = os.path.join(script_dir, 'config.json')  # Construct the absolute path

    if not os.path.exists(config_path):
        print(f"Error: config.json not found at {config_path}. Please create it.")
        exit(1)

    with open(config_path, "r") as f:
        return json.load(f)

# Load configuration
config = get_config()
USERNAME = config["username"]
PASSWORD = config["password"]

AUTH_URL = "https://auth.anaplan.com/token/authenticate"
CLOUDWORKS_API_BASE = "https://api.cloudworks.anaplan.com/2/0/integrations"

def authenticate():
    """Authenticate with Anaplan."""
    auth_string = f"{USERNAME}:{PASSWORD}"
    encoded_auth = base64.b64encode(auth_string.encode()).decode()

    headers = {"Authorization": f"Basic {encoded_auth}", "Content-Type": "application/json"}
    response = requests.post(AUTH_URL, headers=headers)
    
    if response.status_code == 201:
        return response.json()["tokenInfo"]["tokenValue"]
    else:
        print(f"Authentication failed: {response.text}")
        return None

def fetch_integrations(token):
    """Retrieve all integration IDs with schedules and display in a table."""
    headers = {"Authorization": f"AnaplanAuthToken {token}"}
    response = requests.get(CLOUDWORKS_API_BASE, headers=headers)

    if response.status_code == 200:
        data = response.json()
        
        # Extract integration details
        integrations = []
        for integration in data.get("integrations", []):
            if "schedule" in integration:
                integrations.append({
                    "Integration ID": integration["integrationId"],
                    "Integration Name": integration.get("name", "N/A"),
                    "Schedule Time": integration["schedule"].get("time", "N/A"),
                    "Schedule Type": integration["schedule"].get("type", "N/A"),
                    "Timezone": integration["schedule"].get("timezone", "N/A")
                })

        if integrations:
            df = pd.DataFrame(integrations)
            print(df.to_string(index=False))  # Print formatted table output

            # Save to CSV for easy reference
            df.to_csv("cloudworks_integrations.csv", index=False)
            print("\nSaved integration list to 'cloudworks_integrations.csv'")
        else:
            print("No integrations with schedules found.")
    else:
        print(f"Failed to fetch integrations: {response.text}")

# Run the script
token = authenticate()
if token:
    fetch_integrations(token)

Run the Script

python get_cloudworks_integrations.py

Output Example

Integration ID         Integration Name           Schedule Time   Schedule Type   Timezone
ebb5fc3e75f246b6a9f8323ab41b7eda  S3 Export Test         09:09          daily          America/Chicago
a34df6532a7d452d8d123a72c7b5edc2  Data Load Integration  14:30          weekly         America/New_York

The script saves the results to cloudworks_integrations.csv for easy reference.

Step 3: Configure config.json

Update config.json with your Anaplan credentials, integration IDs, and shift settings.

Example config.json File

{
    "username": "email@anaplan.com",
    "password": "your_password",
    "default_timezone": "America/Chicago",
    "shift_hours": 1,
    "integrations": [
        "ebb5fg54h6fg87h8323ab41b7eda",
        "a34df65dfgds545674d8d123a72c"
    ]
}

• Set shift_hours to 1 to move forward by 1 hour.
• Set shift_hours to -1 to move backward by 1 hour.
• Add only the Integration IDs that need updates.

Step 4: Adjust Integration Schedules

Run adjust_cloudworks_schedules.py to update schedules for all listed integrations.

Example Script

import requests
import json
import os
from datetime import datetime, timedelta
import pytz
import base64

def get_config():
    """ Load settings from config.json file, ensuring correct path resolution. """
    script_dir = os.path.dirname(os.path.realpath(__file__))  # Get the script's directory
    config_path = os.path.join(script_dir, 'config.json')  # Construct the absolute path

    if not os.path.exists(config_path):
        print(f"Error: config.json not found at {config_path}. Please create it.")
        exit(1)

    with open(config_path, "r") as f:
        return json.load(f)

# Load configuration
config = get_config()
USERNAME = config["username"]
PASSWORD = config["password"]
DEFAULT_TIMEZONE = config["default_timezone"]
SHIFT_HOURS = config["shift_hours"]  # +1 for forward, -1 for backward
INTEGRATION_IDS = config["integrations"]

# CloudWorks API Endpoints
AUTH_URL = "https://auth.anaplan.com/token/authenticate"
CLOUDWORKS_API_BASE = "https://api.cloudworks.anaplan.com/2/0"

def authenticate():
    """Authenticate with Anaplan and retrieve an access token."""
    auth_string = f"{USERNAME}:{PASSWORD}"
    encoded_auth = base64.b64encode(auth_string.encode()).decode()

    headers = {
        "Authorization": f"Basic {encoded_auth}",
        "Content-Type": "application/json"
    }

    response = requests.post(AUTH_URL, headers=headers)
    
    if response.status_code == 201:
        token = response.json()["tokenInfo"]["tokenValue"]
        print("Authentication successful.")
        return token
    else:
        print(f"Authentication failed: {response.text}")
        return None

def get_current_schedule(token, integration_id):
    """Retrieve the current schedule for a specific integration."""
    url = f"{CLOUDWORKS_API_BASE}/integrations/{integration_id}"
    headers = {"Authorization": f"AnaplanAuthToken {token}"}

    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        schedule_data = response.json()

        if "integration" in schedule_data and "schedule" in schedule_data["integration"]:
            schedule_info = schedule_data["integration"]["schedule"]
            schedule_time = schedule_info.get("time")
            timezone = schedule_info.get("timezone", DEFAULT_TIMEZONE)

            if schedule_time:
                print(f"Integration {integration_id} - Current Schedule: {schedule_time} (Timezone: {timezone})")
                return schedule_time, timezone, schedule_info
            else:
                print(f"Integration {integration_id} - No schedule time found.")
                return None, None, None
        else:
            print(f"Integration {integration_id} - Unexpected API response format.")
            return None, None, None
    else:
        print(f"Integration {integration_id} - Failed to fetch schedule: {response.text}")
        return None, None, None

def adjust_schedule_time(schedule_time, timezone_str, shift_hours):
    """Adjust the integration schedule by a fixed number of hours."""
    local_tz = pytz.timezone(timezone_str)

    try:
        schedule_dt = datetime.strptime(schedule_time, "%H:%M")
    except ValueError:
        print(f"Invalid time format: {schedule_time}. Expected HH:MM format.")
        return None

    adjusted_time = schedule_dt + timedelta(hours=shift_hours)
    
    return adjusted_time.strftime("%H:%M")

def update_schedule(token, integration_id, new_schedule_time, schedule_details):
    """Update the CloudWorks integration schedule with the new adjusted time."""
    if new_schedule_time is None or schedule_details is None:
        print(f"Integration {integration_id} - Skipping update: Missing schedule details or new schedule time.")
        return
    
    url = f"{CLOUDWORKS_API_BASE}/integrations/{integration_id}/schedule"
    headers = {
        "Authorization": f"AnaplanAuthToken {token}",
        "Content-Type": "application/json"
    }
    
    updated_schedule = schedule_details.copy()
    updated_schedule["time"] = new_schedule_time

    payload = {
        "integrationId": integration_id,
        "schedule": updated_schedule
    }

    response = requests.put(url, headers=headers, json=payload)

    if response.status_code == 200:
        print(f"Integration {integration_id} - Schedule updated successfully.")
    else:
        print(f"Integration {integration_id} - Failed to update schedule: {response.text}")

def adjust_schedules():
    """Main function to adjust schedules for all integrations."""
    token = authenticate()
    
    if not token:
        return
    
    for integration_id in INTEGRATION_IDS:
        current_schedule_time, schedule_timezone, schedule_details = get_current_schedule(token, integration_id)

        if not current_schedule_time or not schedule_details:
            print(f"Integration {integration_id} - No valid schedule found. Skipping.")
            continue

        new_schedule_time = adjust_schedule_time(current_schedule_time, schedule_timezone, SHIFT_HOURS)
        update_schedule(token, integration_id, new_schedule_time, schedule_details)

# Run the script
adjust_schedules()

Run the Script

python adjust_cloudworks_schedules.py

Expected Output

Authentication successful.
Integration ebb5fc3e75f246b6a9f8323ab41b7eda - Current Schedule: 09:09 (America/Chicago)
Integration ebb5fc3e75f246b6a9f8323ab41b7eda - Adjusting schedule forward by 1 hour.
Integration ebb5fc3e75f246b6a9f8323ab41b7eda - Schedule updated successfully.

Integration a34df6532a7d452d8d123a72c7b5edc2 - Current Schedule: 14:30 (America/New_York)
Integration a34df6532a7d452d8d123a72c7b5edc2 - Adjusting schedule forward by 1 hour.
Integration a34df6532a7d452d8d123a72c7b5edc2 - Schedule updated successfully.

Frequently Asked Questions (FAQ)

1. Does this script detect if DST is active?

• No. This solution is manually triggered by customers who already know whether they need to shift schedules.

2. Can I run this multiple times?

• Yes! The script only updates the schedule time and preserves all other settings.

3. What if I don’t want to change all integrations?

• Simply edit config.json and remove any integrations you don’t want to adjust.

4. Can I reverse the change if I make a mistake?

• Yes! Just run the script again with shift_hours: -1 (or 1 if you initially moved backward).

Conclusion

This lightweight, easy-to-use Python solution allows CloudWorks customers to bulk adjust their scheduled integrations by +1 or -1 hour.

Got feedback on this content? Let us know in the comments below.

Author: Jon Ferneau, Data Integration Principal, Operational Excellence Group (OEG)

Please note: The articles in our Best Practices section are authored by Anaplan internal experts to provide thought leadership, technical guidance, and educational tutorials. Because every customer environment is unique, these best practices, including any custom code, scripts, or formulas, are provided "as-is" and may not apply directly to your specific use case. Customers are responsible for testing and adapting these solutions for their own environments. Please note that standard Anaplan Support does not cover the troubleshooting, maintenance, or custom modification of code or implementations derived from these articles.

BruceSonos

If we use OAuth to call API, how can we build this "authenticate()" function? Thank you.