Python 重构此函数以将其认知复杂度从 20 降低到允许的 15

Python Refactor this function to reduce its Cognitive Complexity from 20 to the 15 allowed

在 Vscode 中,我看到了来自 sonarlint 的这条消息,并试图弄清楚如何降低此功能的认知复杂性。提前感谢任何帮助。

代码如下:

def job_status(service, headers, job_id):
    """This is my function to look at the Job Status"""
    job_endpoint = service + "Job/" + job_id
    completed = False
    maxtime = 600  # 10 min
    wait = 60
    loop = 0
    logger.info("Endpoint to load Job is : " + job_endpoint)
    while not completed and loop < maxtime:
        r = requests.get(job_endpoint, headers=headers)

        if r.status_code == 200:
            client_resp = r.json()
            if client_resp['jobs'][0] and client_resp['jobs'][0]['jobSummary']:
                current_status = client_resp['jobs'][0]['jobSummary']['status']
                if re.match('^Completed', current_status):
                    completed = True
                elif re.match('^(Kill|Failed|Interrupt)', current_status):  # already dead
                    break
                else:
                    sleep(wait)
                    loop += wait
        else:
            sleep(wait)
            loop += wait

    if not completed:
        if maxtime > loop:
            logger.info("Job failed with status " + current_status)
        else:
            logger.info("Job not completed in " + maxtime + "s with last status " + current_status)
        logger.info("failed")
        sys.exit(5)
    else:
        logger.info("executed successfully")
        sys.exit(0)

我将它分成两个函数,一个用于执行循环,您可以提前-return 退出,另一个用于从那里处理日志记录。

import re
import time
from typing import Tuple, Any

import requests


def _wait_for_status(url, headers, max_time, wait) -> Tuple[str, Any]:
    start_time = time.time()
    current_status = None
    while True:
        if time.time() - start_time > max_time:
            return ("timeout", current_status)

        r = requests.get(url, headers=headers)

        if r.status_code == 200:
            client_resp = r.json()
            first_job = client_resp["jobs"][0]
            if first_job and first_job["jobSummary"]:
                current_status = first_job["jobSummary"]["status"]
                if re.match("^Completed", current_status):
                    return ("success", current_status)
                elif re.match("^(Kill|Failed|Interrupt)", current_status):  # already dead
                    return ("dead", current_status)
        time.sleep(wait)


def job_status(service, headers, job_id):
    """This is my function to look at the Job Status"""
    job_endpoint = service + "Job/" + job_id
    maxtime = 600  # 10 min
    wait = 60
    logger.info("Endpoint to load Job is : " + job_endpoint)
    reason, current_status = _wait_for_status(job_endpoint, headers, maxtime, wait)

    if reason == "timeout":
        logger.info("Exhausted maximum time %s, current status %s", maxtime, current_status)
        sys.exit(5)
    elif reason == "dead":
        logger.info("Failed, current status %s", maxtime, current_status)
        sys.exit(5)
    else:
        logger.info("executed successfully")
        sys.exit(0)