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: <https://gitlab.freedesktop.org/mesa/mesa/-/merge_requests/32637>
This commit is contained in:
Deborah Brouwer
2024-12-13 16:51:07 -08:00
committed by Deb_1543
parent 9671de71a3
commit cb74034517
2 changed files with 33 additions and 24 deletions
+16 -9
View File
@@ -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(<span style='color: {status_color}'>{job.status}</span>,<a href='{job.web_url}'>{job.id}</a>)"
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."
)
+17 -15
View File
@@ -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)