From cb7403451768e67b1bd74c0a951b138dc640cf5b Mon Sep 17 00:00:00 2001 From: Deborah Brouwer Date: Fri, 13 Dec 2024 16:51:07 -0800 Subject: [PATCH] ci: add some static typing to the gantt scripts Add some static typing where possible so that tools like mypy can be used to avoid any future code errors. Part-of: --- bin/ci/ci_gantt_chart.py | 25 ++++++++++++++++--------- bin/ci/ci_post_gantt.py | 32 +++++++++++++++++--------------- 2 files changed, 33 insertions(+), 24 deletions(-) diff --git a/bin/ci/ci_gantt_chart.py b/bin/ci/ci_gantt_chart.py index 8b9472217a0..ddebde55931 100755 --- a/bin/ci/ci_gantt_chart.py +++ b/bin/ci/ci_gantt_chart.py @@ -8,11 +8,16 @@ import argparse +from datetime import datetime, timedelta, timezone +from typing import Dict, List + import gitlab import plotly.express as px -from gitlab_common import pretty_duration -from datetime import datetime, timedelta -from gitlab_common import read_token, GITLAB_URL, get_gitlab_pipeline_from_url +import plotly.graph_objs as go +from gitlab import Gitlab, base +from gitlab.v4.objects import ProjectPipeline +from gitlab_common import (GITLAB_URL, get_gitlab_pipeline_from_url, + pretty_duration, read_token) def calculate_queued_at(job): @@ -35,12 +40,14 @@ def calculate_time_difference(time1, time2): return pretty_duration(diff.seconds) -def create_task_name(job): +def create_task_name(job) -> str: status_color = {"success": "green", "failed": "red"}.get(job.status, "grey") return f"{job.name}\t({job.status},{job.id})" -def add_gantt_bar(job, tasks): +def add_gantt_bar( + job: base.RESTObject, tasks: List[Dict[str, str | datetime | timedelta]] +) -> None: queued_at = calculate_queued_at(job) task_name = create_task_name(job) @@ -73,12 +80,12 @@ def add_gantt_bar(job, tasks): ) -def generate_gantt_chart(pipeline): +def generate_gantt_chart(pipeline: ProjectPipeline): if pipeline.yaml_errors: raise ValueError("Pipeline YAML errors detected") # Convert the data into a list of dictionaries for plotly - tasks = [] + tasks: List[Dict[str, str | datetime | timedelta]] = [] for job in pipeline.jobs.list(all=True, include_retried=True): add_gantt_bar(job, tasks) @@ -94,7 +101,7 @@ def generate_gantt_chart(pipeline): ) # Create a Gantt chart - fig = px.timeline( + fig: go.Figure = px.timeline( tasks, x_start="Start", x_end="Finish", @@ -125,7 +132,7 @@ def generate_gantt_chart(pipeline): return fig -def parse_args() -> None: +def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Generate the Gantt chart from a given pipeline." ) diff --git a/bin/ci/ci_post_gantt.py b/bin/ci/ci_post_gantt.py index 2a00a2a79d6..04f7939e1cc 100755 --- a/bin/ci/ci_post_gantt.py +++ b/bin/ci/ci_post_gantt.py @@ -8,26 +8,27 @@ import argparse -import gitlab -import re import logging as log import os -import pytz +import re import traceback from datetime import datetime, timedelta -from gitlab_common import ( - read_token, - GITLAB_URL, - get_gitlab_pipeline_from_url, -) +from typing import Any, Dict + +import gitlab +import pytz from ci_gantt_chart import generate_gantt_chart +from gitlab import Gitlab +from gitlab.base import RESTObject +from gitlab.v4.objects import Project, ProjectPipeline +from gitlab_common import GITLAB_URL, get_gitlab_pipeline_from_url, read_token MARGE_USER_ID = 9716 # Marge LAST_MARGE_EVENT_FILE = os.path.expanduser("~/.config/last_marge_event") -def read_last_event_date_from_file(): +def read_last_event_date_from_file() -> str: try: with open(LAST_MARGE_EVENT_FILE, "r") as f: last_event_date = f.read().strip() @@ -37,7 +38,7 @@ def read_last_event_date_from_file(): return last_event_date -def pretty_time(time_str): +def pretty_time(time_str: str) -> str: """Pretty print time""" local_timezone = datetime.now().astimezone().tzinfo @@ -47,7 +48,7 @@ def pretty_time(time_str): return f'{time_str} ({time_d.strftime("%d %b %Y %Hh%Mm%Ss")} {local_timezone})' -def compose_message(file_name, attachment_url): +def compose_message(file_name: str, attachment_url: str) -> str: return f""" Here is the Gantt chart for the referred pipeline, I hope it helps 😄 (tip: click on the "Pan" button on the top right bar): @@ -61,13 +62,13 @@ This message was generated by the ci_post_gantt.py script, which is running on a """ -def gitlab_upload_file_get_url(gl, project_id, filepath): - project = gl.projects.get(project_id) - uploaded_file = project.upload(filepath, filepath=filepath) +def gitlab_upload_file_get_url(gl: Gitlab, project_id: str, filepath: str) -> str: + project: Project = gl.projects.get(project_id) + uploaded_file: Dict[str, Any] = project.upload(filepath, filepath=filepath) return uploaded_file["url"] -def gitlab_post_reply_to_note(gl, event, reply_message): +def gitlab_post_reply_to_note(gl: Gitlab, event: RESTObject, reply_message: str): """ Post a reply to a note in thread based on a GitLab event. @@ -158,6 +159,7 @@ if __name__ == "__main__": try: log.info(f"Found message: {event.note['body']}") pipeline_url = match.group(0)[:-1] + pipeline: ProjectPipeline pipeline, _ = get_gitlab_pipeline_from_url(gl, pipeline_url) log.info("Generating gantt chart...") fig = generate_gantt_chart(pipeline)