Adjusting CloudWorks Integration Schedules for Daylight Saving Time (DST) Using Python
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:
- get_cloudworks_integrations.py – Retrieves a list of integrations with schedules
- config.json – Stores authentication details and integration IDs
- 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)
Tagged:
2
Comments
-
If we use OAuth to call API, how can we build this "authenticate()" function? Thank you.
0 -
@BruceSonos Here is a guide on implementing OAuth in Python:
0
