Блог

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

Я бы разделил это на две функции: одну, которая выполняет цикл, из которого вы можете досрочно вернуться, и другую, которая занимается протоколированием оттуда.

 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)