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

Question

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.

• restapi.py(86, 5): +1
• restapi.py(86, 25): +1
• restapi.py(89, 9): +2 (incl 1 for nesting)
• restapi.py(91, 13): +3 (incl 2 for nesting)
• restapi.py(91, 39): +1
• restapi.py(93, 17): +4 (incl 3 for nesting)
• restapi.py(95, 17): +1
• restapi.py(97, 17): +1
• restapi.py(100, 9): +1
• restapi.py(104, 5): +1
• restapi.py(105, 9): +2 (incl 1 for nesting)
• restapi.py(107, 9): +1
• restapi.py(111, 5): +1

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)

Answer 1

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)