Hi @MekakundeNithin ,
I have done something similar using python script. I am attaching the script hope it is helpful to you.This is only for source config object but you can add whatever you want.
import glob
import http.client
import json
import shutil
import sys
import time
from datetime import datetime
from pathlib import Path
from urllib.parse import urlencode, urlparse
EXPORT_OBJECT_TYPE = "SOURCE"
POLL_INTERVAL_SECONDS = 15
EXPORT_TIMEOUT_SECONDS = 3600
OUTPUT_PREFIX = "sp_config_sources_export"
ARCHIVE_FOLDER = "archive"
IGNORE_FAILED_EXPORT = True
class SailPointClient:
def __init__(self, tenant, client_id, client_secret, token_url):
self.tenant = self._normalize_tenant(tenant)
self.api_base_url = f"https://{self.tenant}.api.identitynow.com"
self.client_id = client_id
self.client_secret = client_secret
self.token_url = token_url
self.access_token = None
self.token_expiry_epoch = 0
def _normalize_tenant(self, value):
value = value.strip()
value = value.replace("https://", "").replace("http://", "").strip("/")
if value.endswith(".api.identitynow.com"):
value = value[:-len(".api.identitynow.com")]
return value
def _build_path(self, parsed):
path = parsed.path or "/"
if parsed.query:
path += "?" + parsed.query
return path
def _decode_bytes(self, data):
if not data:
return ""
return data.decode("utf-8", errors="replace")
def _json_or_text(self, data):
text = self._decode_bytes(data)
if not text:
return {}
try:
return json.loads(text)
except Exception:
return {"raw_text": text}
def _request_raw(self, method, url, headers=None, body=None, retries=6):
headers = headers or {}
parsed = urlparse(url)
if parsed.scheme != "https":
raise ValueError(f"Only https URLs are supported: {url}")
host = parsed.netloc
path = self._build_path(parsed)
last_status = None
last_headers = {}
last_data = b""
for attempt in range(1, retries + 1):
conn = None
try:
conn = http.client.HTTPSConnection(host, timeout=120)
conn.request(method, path, body=body, headers=headers)
response = conn.getresponse()
status = response.status
response_headers = {k.lower(): v for k, v in response.getheaders()}
data = response.read()
last_status = status
last_headers = response_headers
last_data = data
if status == 429:
retry_after = response_headers.get("retry-after", "")
wait_time = int(retry_after) if retry_after.isdigit() else min(60, attempt * 2)
print(f"429 received. Waiting {wait_time}s and retrying...")
time.sleep(wait_time)
continue
if 500 <= status < 600 and attempt < retries:
wait_time = min(60, attempt * 2)
print(f"Server error {status}. Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
return status, response_headers, data
except Exception as e:
if attempt < retries:
wait_time = min(60, attempt * 2)
print(f"Request failed: {e}. Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
raise
finally:
if conn:
conn.close()
return last_status, last_headers, last_data
def get_token(self, force=False):
now = time.time()
if not force and self.access_token and now < self.token_expiry_epoch - 60:
return self.access_token
body = urlencode({
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
})
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json"
}
status, _, data = self._request_raw(
"POST",
self.token_url,
headers=headers,
body=body
)
if status < 200 or status >= 300:
raise RuntimeError(
f"Failed to get token. HTTP {status}. Response: {self._decode_bytes(data)}"
)
token_json = self._json_or_text(data)
access_token = token_json.get("access_token")
expires_in = int(token_json.get("expires_in", 3600))
if not access_token:
raise RuntimeError(f"Token response missing access_token: {token_json}")
self.access_token = access_token
self.token_expiry_epoch = now + expires_in
return self.access_token
def request_json(self, method, url, json_body=None, retry_on_401=True):
self.get_token()
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {self.access_token}"
}
body = None
if json_body is not None:
body = json.dumps(json_body)
headers["Content-Type"] = "application/json"
status, response_headers, data = self._request_raw(
method,
url,
headers=headers,
body=body
)
if status == 401 and retry_on_401:
print("401 received. Refreshing token and retrying...")
self.get_token(force=True)
headers["Authorization"] = f"Bearer {self.access_token}"
status, response_headers, data = self._request_raw(
method,
url,
headers=headers,
body=body
)
return status, response_headers, self._json_or_text(data)
def download_file(self, url, output_file, retry_on_401=True):
self.get_token()
headers = {
"Accept": "*/*",
"Authorization": f"Bearer {self.access_token}"
}
status, response_headers, data = self._request_raw(
"GET",
url,
headers=headers
)
if status == 401 and retry_on_401:
print("401 received during download. Refreshing token and retrying...")
self.get_token(force=True)
headers["Authorization"] = f"Bearer {self.access_token}"
status, response_headers, data = self._request_raw(
"GET",
url,
headers=headers
)
if status < 200 or status >= 300:
raise RuntimeError(
f"Failed to download export result. HTTP {status}. Response: {self._decode_bytes(data)}"
)
with open(output_file, "wb") as f:
f.write(data)
return response_headers
def load_creds(path="creds.json"):
creds_path = Path(path)
if not creds_path.exists():
raise FileNotFoundError(f"Credentials file not found: {path}")
with open(creds_path, "r", encoding="utf-8") as f:
creds = json.load(f)
tenant = creds.get("tenant") or creds.get("domain")
client_id = creds.get("clientid") or creds.get("client_id")
client_secret = creds.get("clientsecret") or creds.get("client_secret")
token_url = creds.get("token_url") or creds.get("token_endpoint")
if not tenant:
raise ValueError("Missing 'tenant' or 'domain' in creds.json")
if not client_id:
raise ValueError("Missing 'clientid' or 'client_id' in creds.json")
if not client_secret:
raise ValueError("Missing 'clientsecret' or 'client_secret' in creds.json")
if not token_url:
raise ValueError("Missing 'token_url' or 'token_endpoint' in creds.json")
return {
"tenant": tenant,
"client_id": client_id,
"client_secret": client_secret,
"token_url": token_url
}
def archive_old_export_files():
archive_dir = Path(ARCHIVE_FOLDER)
archive_dir.mkdir(exist_ok=True)
files_moved = 0
for file_path in glob.glob(f"{OUTPUT_PREFIX}_*.json"):
src = Path(file_path)
if not src.is_file():
continue
dst = archive_dir / src.name
if dst.exists():
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
dst = archive_dir / f"{src.stem}_{stamp}{src.suffix}"
shutil.move(str(src), str(dst))
files_moved += 1
print(f"Moved old export file to {dst}")
if files_moved == 0:
print("No old source export files found to archive.")
def generate_output_filename():
timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
return f"{OUTPUT_PREFIX}_{timestamp}.json"
def build_export_payload():
description = f"Weekly SP-Config SOURCE Export - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
return {
"description": description,
"includeTypes": [
EXPORT_OBJECT_TYPE
]
}
def extract_export_job_id(response_data):
if not isinstance(response_data, dict):
raise RuntimeError(f"Unexpected export response: {response_data}")
for field in ["id", "jobId", "exportJobId"]:
value = response_data.get(field)
if value:
return value
raise RuntimeError(f"Could not find export job ID in response: {response_data}")
def start_sp_config_source_export(client):
url = f"{client.api_base_url}/v2025/sp-config/export"
payload = build_export_payload()
print("Starting SP-Config SOURCE export job...")
print(f"Export payload: {json.dumps(payload, indent=2)}")
status, _, data = client.request_json(
"POST",
url,
json_body=payload
)
if status < 200 or status >= 300:
raise RuntimeError(
f"Failed to start SOURCE export. HTTP {status}. Response: {data}"
)
export_job_id = extract_export_job_id(data)
print(f"SOURCE export job started. Job ID: {export_job_id}")
return export_job_id
def get_sp_config_export_status(client, export_job_id):
url = f"{client.api_base_url}/v2025/sp-config/export/{export_job_id}"
status, _, data = client.request_json("GET", url)
if status < 200 or status >= 300:
raise RuntimeError(
f"Failed to get export job status. HTTP {status}. Response: {data}"
)
return data
def normalize_job_status(status_data):
if not isinstance(status_data, dict):
return "UNKNOWN"
for field in ["status", "state", "jobStatus"]:
value = status_data.get(field)
if value:
return str(value).upper()
if status_data.get("completed") is True:
return "COMPLETED"
return "UNKNOWN"
def poll_until_export_complete(client, export_job_id):
start_time = time.time()
success_statuses = {
"COMPLETE",
"COMPLETED",
"SUCCESS",
"SUCCEEDED",
"FINISHED"
}
failure_statuses = {
"FAILED",
"FAILURE",
"ERROR",
"CANCELLED",
"CANCELED",
"FAILED_EXTERNAL_COMMUNICATION"
}
while True:
elapsed = time.time() - start_time
if elapsed > EXPORT_TIMEOUT_SECONDS:
message = f"SOURCE export job did not complete within {EXPORT_TIMEOUT_SECONDS} seconds."
if IGNORE_FAILED_EXPORT:
print(f"WARNING: {message}")
return False
raise TimeoutError(message)
status_data = get_sp_config_export_status(client, export_job_id)
job_status = normalize_job_status(status_data)
print(f"SOURCE export job status: {job_status}")
if job_status in success_statuses:
print("SOURCE export job completed successfully.")
return True
if job_status in failure_statuses:
print("WARNING: SOURCE export job failed.")
print(f"Failure details: {json.dumps(status_data, indent=2)}")
if IGNORE_FAILED_EXPORT:
print("Skipping SOURCE export because it failed.")
return False
raise RuntimeError(f"SOURCE export job failed. Status response: {status_data}")
time.sleep(POLL_INTERVAL_SECONDS)
def download_sp_config_export(client, export_job_id, output_file):
url = f"{client.api_base_url}/v2025/sp-config/export/{export_job_id}/download"
print(f"Downloading SOURCE export result to {output_file}...")
response_headers = client.download_file(url, output_file)
print(f"Download completed successfully: {output_file}")
content_type = response_headers.get("content-type", "")
if content_type:
print(f"Download content type: {content_type}")
def main():
try:
creds = load_creds("creds.json")
archive_old_export_files()
client = SailPointClient(
tenant=creds["tenant"],
client_id=creds["client_id"],
client_secret=creds["client_secret"],
token_url=creds["token_url"]
)
client.get_token()
export_job_id = start_sp_config_source_export(client)
completed = poll_until_export_complete(client, export_job_id)
if not completed:
print("SOURCE export was skipped because the export job failed or timed out.")
sys.exit(0)
output_file = generate_output_filename()
download_sp_config_export(client, export_job_id, output_file)
print("SP-Config SOURCE export backup finished successfully.")
except Exception as e:
print(f"ERROR: {e}")
sys.exit(1)
if __name__ == "__main__":
main()