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)
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)