python 从Jenkins API获取所有构建日志

xqkwcwgp  于 2023-02-02  发布在  Python
关注(0)|答案(1)|浏览(261)

我需要从Jenkins示例中获取所有构建版本的日志。

def get_builds():
    builds_api_res = session.get('https://build.org.com/api/json?depth=3',stream=True,auth=(username,access_token),).json()
    for chunk in builds_api_res.iter_content(chunk_size=1024*1024):
        # Do something with the chunk of data
        print(chunk)
get_builds()

问题是,这返回了如此巨大的响应,以至于Jenkins示例本身耗尽了内存。
因此,另一种方法是从每个文件夹中单独获取所有构建版本,这就是我面临的问题。
当我查看文件夹时,一些 * 构建 * 位于2个文件夹级别之下,一些位于3个级别之下,也许还有从根文件夹向下4个文件夹级别的构建。我不知道如何继续在文件夹中查找文件夹,除非我找到一个构建。

for project in projects_api['jobs']:
    folder_url = JENKINS_URL+'/job/'+project['name']+'/api/json'
    folders = session.get(folder_url, auth=(username, access_token)).json()
    jobs = folders['jobs']
    for job in jobs:
        job_det_url = job['url'] + 'api/json'
        all_builds = session.get(job_det_url, auth=(username, access_token)).json()
        latest_build = all_builds['builds'][-1]
        print(latest_build['url'])
        log_url = latest_build['url'] +'/consoleText'
        print(log_url)
        logs = session.get(log_url, auth=(username, access_token))
        print(logs.text)

这适用于 root 文件夹中的构建版本。但是,如果该文件夹中有更多的文件夹,则它将失败。如果有办法,我如何使用API* 直接从父文件夹定位构建版本?或者有更好的方法来完成整个过程?

6g8kf2rb

6g8kf2rb1#

我决定在一个循环中继续检查文件夹中的文件夹,如果发现构建,就跳出这个循环。

def check_builds(job):
    try:
        if job['color']:
            print("Build found, sleepng for 7 seonds")
            time.sleep(7)
            build_url = job['url'] + 'api/json'
            builds = session.get(build_url, auth=(username,access_token),stream=True).json()
            resp = builds['builds'][0]['url']
            log_url = resp + 'consoleText'
            time.sleep(10)
            print("Logs Found. Sleeping for 10 seconds")
            logs = session.get(log_url, auth=(username,access_token),stream=True)
            print(logs.text)
            return True
    except:
        return job['url']

def no_build(job):
    print("No builds found. Checking folders in 7 seonds")
    time.sleep(7)
    next_url = job + 'api/json'
    print(next_url)
    try:
        next_level = session.get(next_url, auth=(username,access_token),stream=True).json()
        for job in next_level['jobs']:
            resp = check_builds(job)
            if resp is not True:
                no_build(resp)
    except:
        pass

# Get a list of all jobs from root folders
projects_api = session.get(JENKINS_URL + '/api/json', auth=(username,access_token),stream=True).json()
for root_folder in projects_api['jobs']:
    root_folder_job_url = JENKINS_URL + 'job/'+ root_folder['name']+ '/api/json/tree=jobs[name,url,builds[number,url]]'
    root_jobs = session.get(root_folder_job_url, auth=(username,access_token),stream=True).json()
    for r_job in root_jobs['jobs']:
        check = check_builds(r_job)
        if check is not True:
            no_build(check)

相关问题