def main()

in cdk-project/lib/images/codebuild-image/python/src/notebooks/cli/describe_notebook_jobs.py [0:0]
81 lines of code
11 McCabe index (conditional complexity)

def main():
    args = parse_args(sys.argv[1:])

    session = ensure_session()

    csv_filename = args.csv
    dataframe = pd.read_csv(csv_filename, index_col=False)

    output_notebooks = []
    runtimes = []
    statuses = []
    errors = []
    dates = []
    error_details= []

    sagemaker = session.client("sagemaker")
    for index, row in dataframe.iterrows():
        job_name = row["processing-job-name"]
        detail = None
        if job_name == "None":
            uri = "None"
            runtime = 0
            status = "Skipped"
            error = None
            date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        else:
            response = sagemaker.describe_processing_job(ProcessingJobName=job_name)
            date = response.get("ProcessingEndTime", datetime.now(timezone.utc)).strftime(
                "%Y-%m-%d"
            )
            notebook, uri = get_output_notebook(job_name, session)
            status = response.get("ProcessingJobStatus")

            runtime = (
                response.get("ProcessingEndTime", datetime.now(timezone.utc))
                - response.get("ProcessingStartTime", datetime.now(timezone.utc))
            ).total_seconds()
            if runtime < 0:
                runtime = 0

            error = response.get("ExitMessage")
            if error == "Kernel died":
                error = "KernelDied"
                detail = "kernel died"
            elif error:
                found_error_type = False
                valid_error_types = ("Exception:", "Error:", "InvalidArn:", "NotFound:", "InUse:")
                lines = error.splitlines()
                for line in reversed(lines):
                    if any(error_type in line for error_type in valid_error_types):
                        error_parsed = line.split(":", 1)
                        print("The following error was encountered while executing the notebook")
                        print(line)
                        error = error_parsed[0]
                        detail = error_parsed[1]
                        found_error_type = True
                        break
                if not found_error_type:
                    error = "Uncategorized"

            if status == "Stopped":
                error = "TimedOut"
                detail = "Notebook execution timed out"

        output_notebooks.append(uri)
        runtimes.append(runtime)
        statuses.append(status)
        errors.append(error)
        dates.append(date)
        error_details.append(detail)

        print(job_name)
        time.sleep(1)

    new_dataframe = pd.DataFrame(
        {
            "date": dates,
            "filename": dataframe["filename"],
            "processing-job-name": dataframe["processing-job-name"],
            "kernel": dataframe["kernel"],
            "output": output_notebooks,
            "runtime": runtimes,
            "status": statuses,
            "error": errors,
            "error_detail": error_details
        }
    )

    print("\n" * 2)
    print("-" * 100)
    print("\n" * 2)
    print(save_csv_to_s3(new_dataframe, csv_filename))