based on https://github.com/linhkikuchi/Delete-Old-Executions-Rundeck/blob/master/deleteOldExecutionFromAPI.py

rundeck_delete_old_executions.py

#!/bin/python
import requests
import xml.etree.ElementTree as ET
import time
import sys
from datetime import datetime

class RundeckDelete:
    API_KEY = ''
    EXPIRE_DAYS = 50
    TODAY = int(round(time.time() * 1000))
    EXPIRE_MILISECONDS = EXPIRE_DAYS * 24 * 60 * 60 * 1000
    VERIFY_REQUEST = True
    baseUrl = ''
    headers = {}

    def __init__(self, rundeckUrl, api_key):
        self.baseUrl = rundeckUrl
        self.API_KEY = api_key
        self.headers = {'Content-Type': 'application/json', 'X-RunDeck-Auth-Token': self.API_KEY}
        pass

    # API call to get the list of the existing projects on the server.
    def listProjects(self):
        url = self.baseUrl + '/api/1/projects'
        r = requests.get(url, headers=self.headers, verify=self.VERIFY_REQUEST)
        if not r:
            r = requests.get(url, headers=self.headers, verify=self.VERIFY_REQUEST)
            r.raise_for_status()
            # raise Exception("can't get list of projects from " + url)
        return r.text.encode('utf-8')

    # Returns list of all the project names
    def getProjectNames(self, projectsinfo_xml):
        project_names = []
        root = ET.fromstring(projectsinfo_xml)
        for projects in root:
            for name in projects.findall('project'):
                project_names.append(name.find('name').text)
        return project_names

    # API call to get the list of the jobs that exist for a project.
    def listJobsForProject(self, project_name):
        url = self.baseUrl+'/api/1/jobs'
        payload = { 'project':  project_name }
        r = requests.get(url, params=payload, headers=self.headers, verify=self.VERIFY_REQUEST)
        if not r:
            r = requests.get(url, params=payload, headers=self.headers, verify=self.VERIFY_REQUEST)
            r.raise_for_status()
            # raise Exception("can't get jobs in project " + project_mame + " from " + url)
        return r.text.encode('utf-8')

    # Returns list of all the jobids
    def getJobIDs(self, jobsinfo_xml):
        job_ids = []
        root = ET.fromstring(jobsinfo_xml)
        for jobs in root:
            for job in jobs:
                job_ids.append(job.attrib['id'])
        return job_ids

    # API call to get the list of the executions for a Job.
    def getExecutionsForAJob(self, job_id):
        url = self.baseUrl+'/api/1/job/'+job_id+'/executions'
        r = requests.get(url, headers=self.headers, verify=self.VERIFY_REQUEST)
        if not r:
            r = requests.get(url, headers=self.headers, verify=self.VERIFY_REQUEST)
            if not r:
                print("cant get executions for job " + job_id)
                return None
                r.raise_for_status()
        # if not r:
        #     raise Exception("can't get executions in job " + job_id + " from " + url)
        return r.text.encode('utf-8')

    # Returns a dict {'execution_id01': 'execution_date01', 'execution_id02': 'execution_date02', ... }
    def getExecutionDate(self, executionsinfo_xml):
        execid_dates = {}
        root = ET.fromstring(executionsinfo_xml)
        for executions in root:
            for execution in executions.findall('execution'):
                execution_id = execution.get('id')
                execution_date = None
                for date in execution.findall('date-ended'):
                    execution_date = date.get('unixtime')
                if execution_date is None:
                    print("No execution_date for " + str(execution_id))
                else:
                    execid_dates[execution_id] = execution_date
        return execid_dates

    #API call to delete an execution by ID
    def deleteExecution(self, execution_id):
        print("delete " + execution_id)
        url = self.baseUrl+'/api/12/execution/'+execution_id
        r = requests.delete(url, headers=self.headers, verify=self.VERIFY_REQUEST)
        if not r:
            #r = requests.delete(url, headers=self.headers, verify=self.VERIFY_REQUEST)
            #r.raise_for_status()
            print("can't delete " + execution_id)
            #raise Exception("can't delete execution " + execution_id + " from " + url)

    def isOlderThanExpireDays(self, execution_date, today):
        if ((today - execution_date) > self.EXPIRE_MILISECONDS):
            return True
        return False

    def checkDeletion(self, execid_dates):
        for exec_id, exec_date in execid_dates.iteritems():
            if self.isOlderThanExpireDays(int(exec_date), self.TODAY):
                print("expired execId=" + exec_id + " exec date=" + str(datetime.utcfromtimestamp(int(exec_date)/1000)))
                self.deleteExecution(exec_id)

    def main(self):
        projects = self.getProjectNames(self.listProjects())
        for project in projects:
            print('project:\t' + project)
            jobids = self.getJobIDs(self.listJobsForProject(project))
            for jobid in jobids:
                print('jobid:\t' + jobid)
                executions_list = self.getExecutionsForAJob(jobid)
                if executions_list is not None:
                    self.checkDeletion(self.getExecutionDate(executions_list))


if __name__ == "__main__":
    print("Removing excutions from Rundeck...")
    import sys
    if '-verbose' in sys.argv or '-debug' in sys.argv:
        try:
            import IPython.core.ultratb
            sys.excepthook = IPython.core.ultratb.FormattedTB(mode='Verbose', color_scheme='Linux',
                                                              call_pdb=1 if '-debug' in sys.argv else 0)
        except ImportError:
            print("Can't import IPython for debug")
            pass
    RundeckDelete(sys.argv[2], sys.argv[1]).main()
    print("Done")