04-03-2025 01:49 AM
Hi Team,
I've developed a Python script to copy Tidal jobs on demand. While the dependencies are visible in the GUI, I'm unable to retrieve them through the API method. Below is my implementation:
For copying a job:
def copy_jobonly(path): jobid, parentid = get_Job_Detail(path) print("Inside copy_jobonly", path) print(jobid, parentid) url = "http://xxxx/api/tes-6.5/post" credentials = f"{username}:{password}" encoded_credentials = base64.b64encode(credentials.encode()).decode() headers = { "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {encoded_credentials}" } post_command = f"""<?xml version="1.0" encoding="UTF-8" ?> <entry xmlns="http://purl.org/atom/ns#"> <tes:Job.copy xmlns:tes="http://www.tidalsoftware.com/client/tesservlet"> <id>{jobid}</id> <parentid>{parentid}</parentid> <prefix>copied_</prefix> </tes:Job.copy> </entry>""" payload = {"data": post_command} try: response = requests.post(url, data=payload, headers=headers) if response.status_code == 200: print("Response:", response.text) return response.text else: print(f"Error: {response.status_code}, Message: {response.text}") except requests.exceptions.RequestException as e: print(f"An error occurred: {e}")
For getting the dependencies:
def get_Alldependencies(): url = "http://xxxx/api/tes-6.5/JobDependency.getList" try: # Make the GET request response = requests.get(url, auth=HTTPBasicAuth(username, password)) response.raise_for_status() # Check if the response is Atom XML if response.headers.get("Content-Type").startswith("application/atom+xml"): root = ET.fromstring(response.text) namespaces = { 'atom': 'http://purl.org/atom/ns#', # Atom namespace 'tes': 'http://www.tidalsoftware.com/client/tesservlet' # TES namespace } # Initialize a list to store the data data = [] # Loop through entries in the XML response for entry in root.findall(".//atom:entry", namespaces=namespaces): job = entry.find(".//tes:jobdependency", namespaces=namespaces) # Extract job details prim_id = job.find(".//tes:id", namespaces=namespaces) jobid = job.find(".//tes:jobid", namespaces=namespaces) jobname = job.find(".//tes:jobname", namespaces=namespaces) depjobname = job.find(".//tes:depjobname", namespaces=namespaces) depjobparent = job.find(".//tes:depjobparent", namespaces=namespaces) depjobid = job.find(".//tes:depjobid", namespaces=namespaces) status = job.find(".//tes:status", namespaces=namespaces) # Append the extracted data as a dictionary try: data.append({ "Primary ID": prim_id.text, "Job ID": jobid.text, "Job Name": jobname.text, "Dependent Job Name": depjobname.text, "Dependent Job Parent": depjobparent.text, "Dependent Job Id": depjobid.text, "Status": status.text }) except: data.append({ "Primary ID": None, "Job ID": None, "Job Name": None, "Dependent Job Name": None, "Dependent Job Parent": None, "Dependent Job Id": None }) print(data[-1]) # Convert the list of dictionaries to a DataFrame df = pd.DataFrame(data) print(df) # Save the DataFrame to a CSV file df.to_csv(r"C:\Users\xxxx\Downloads\job_dependencies.csv", index=False) print("Data has been written to job_dependencies.csv") else: print("Unexpected response format. Expected Atom XML.") except requests.exceptions.RequestException as e: print(f"An error occurred: {e}")
Issue:
Could you please help identify:
I've attached the complete code for reference.
Thank you for your assistance.
Discover and save your favorite ideas. Come back to expert answers, step-by-step guides, recent topics, and more.
New here? Get started with these tips. How to use Community New member guide