Python 重构此函数以将其认知复杂度从 20 降低到允许的 15
Python Refactor this function to reduce its Cognitive Complexity from 20 to the 15 allowed
在 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)
我将它分成两个函数,一个用于执行循环,您可以提前-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)
在 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)
我将它分成两个函数,一个用于执行循环,您可以提前-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)