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)