pythonpython-3.xsonarlintcyclomatic-complexity

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


In Vscode, I see this message from sonarlint and trying to figure out how to reduce the Cognitive Complexity of this function. Any assistance is appreciated in advance.

Here is the code :

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)

Solution

  • I'd split this into two functions, one that does the looping you can early-return out of, and the other that deals with logging from there.

    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)