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

Question

在Vscode中，我看到了来自sonarlint的这条消息，并试图找出如何降低该函数的认知复杂性。如有任何协助，我们将不胜感激。

• restapi.py(86，5(：+1
• restapi.py(86，25(：+1
• restapi.py(89，9(：+2(包括1用于嵌套(
• restapi.py(91，13(：+3(包括嵌套的2(
• restapi.py(91，39(：+1
• restapi.py(93，17(：+4(包括嵌套的3(
• 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(包括1用于嵌套(
• restapi.py(107，9(：+1
• restapi.py(111，5(：+1

这是代码：

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

我将其分为两个函数，一个用于执行可以提前返回的循环，另一个用于处理日志记录。

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)