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)