Adjusting CloudWorks Integration Schedules for Daylight Saving Time (DST) 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 Daylight Saving Time (DST) changes or other 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

Anaplan CloudWorks schedules integrations based on Coordinated Universal Time (UTC). In regions observing DST, local times shift by an hour during specific periods, causing integrations to run at unintended times if not adjusted. Automating these adjustments ensures integrations align with local business hours throughout the year.

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)

BruceSonos

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

JonFerneau

@BruceSonos Here is a guide on implementing OAuth in Python:

OEG Best Practice: Anaplan Python OAuth Example

AnaplanOEG

Feb 15, 2023

The Anaplan Python OAuth Example demonstrates using the Anaplan REST API with OAuth with device-based authorization. The code highlights how to generate a device_id, access_token, and refresh_token. Additionally, the code highlights a multi-threaded approach to request a new access_token while performing other, longer-running operations such as a large data load. Please note that with this code example, the concept is simulated by calling Get Workspaces five times every 10 seconds while simultaneously refreshing the access_token every 5 seconds. Please note that the access_token expires after 35 minutes, so please adjust the timer to a value just below this expiration length (2100 seconds). Usage and the source code can be viewed here.

Here is a brief overview of the example:

1) New device-based Oauth Registration:

https://us.v-cdn.net/6037036/uploads/D29080P13JAT/anaplan-oauth-token-refresh-new-device-registration.gif

2) After a successful registration, run unattended mode:

https://us.v-cdn.net/6037036/uploads/W0K9Q8SAEHBO/anaplan-oauth-token-refresh-device-registered.gif

3) To see all command line arguments, type: 'python3 anaplan.py -h'

https://us.v-cdn.net/6037036/uploads/D0PBKINNF905/anaplan-oauth-help.gif

4) To update any of the Anaplan API URLs, please edit the file settings.json.

Note: The client_id and refresh_token are stored as encrypted values in a SQLite database. As an alternative, a solution like auth0 would further enhance security.

Author: Quin Eddy, @QuinE - Director of Data Integration, Operational Excellence Group (OEG)

---

View Post