14861206140

Committed 06 May 2025 01:41PM UTC coverage: 96.771% (-0.4%) from 97.128%

Build # 14861206140

Build Type

push

github

Committed by

web-flow

Commit Message

Gitlab presentation task (#95)

* Gitlab presentation task

- creates issues on git.smhi.se
- file upload, issue creation, and issue modification work as with Redmine
- modified create_anchor() to work with Gitlab
- added new dependency for python-gitlab
- added some tests for the new task
- Update documentation for Gitlab task

* Reordered presentation task documentation

- ordered tasks alphabetically (Gitlab is probably most used)
- moved custom visualization options to the end (closes #104)

* Update CHANGES

* More explicit server connection treatment

- make the URL and project ID module constants as suggested by @uwefladrich
- the `gl` object was not used outside of `get_project_and_issue`, so I moved its creation into the function.
This way, the repo information are used in the same place, which is easier to understand I think

Run Details

125 of 137 new or added lines in 2 files covered. (91.24%)

2188 of 2261 relevant lines covered (96.77%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.21

/monitoring/gitlab.py

"""Presentation Task that uploads Data and Plots to the EC-Earth 4 Gitlab."""

import re
import urllib.parse
from pathlib import Path

import gitlab
import jinja2
import requests.exceptions
from scriptengine.exceptions import ScriptEngineTaskRunError
from scriptengine.jinja import filters as j2filters
from scriptengine.tasks.core import Task, timed_runner

from helpers.exceptions import PresentationException
from helpers.presentation_objects import PresentationObject

# Repository where monitoring results are posted
SERVER_URL = "https://git.smhi.se"
PROJECT_ID = 1982


class Gitlab(Task):
    """Gitlab Presentation Task"""

    _required_arguments = (
        "src",
        "local_dst",
        "subject",
        "template",
        "api_key",
    )

    def __init__(self, arguments=None):
        Gitlab.check_arguments(arguments)
        super().__init__(arguments)

    @timed_runner
    def run(self, context):
        sources = self.getarg("src", context)
        dst_folder = self.getarg("local_dst", context)
        issue_subject = self.getarg("subject", context, parse_yaml=False)
        template_path = self.getarg("template", context)
        key = self.getarg("api_key", context)
        self.log_info(f"Create Gitlab issue '{issue_subject}'.")
        self.log_debug(f"Template: {template_path}, Source File(s): {sources}")

        presentation_list = self.get_presentation_list(sources, dst_folder)
        gitlab_template = self.get_template(context, template_path)
        gitlab_template.globals["urlencode"] = urllib.parse.quote

        self.log_debug("Connecting to Gitlab.")
        project, issue = self.get_project_and_issue(key, issue_subject)

        self.log_debug("Uploading attachments.")
        issue.uploads = []
        for item in presentation_list:
            if item["presentation_type"] == "image":
                file_name = Path(item["path"]).name
                upload = project.upload(file_name, filepath=f"{dst_folder}/{file_name}")
                item["image_link"] = upload[
                    "markdown"
                ]  # string of the form "![]()" generated by Gitlab

        # render the template and add as description
        self.log_debug("Updating the issue description.")
        issue.description = gitlab_template.render(
            presentation_list=presentation_list,
            issue_url=issue.web_url,
            create_anchor=create_anchor,
        )
        self.log_debug("Saving issue.")
        issue.save()

    def get_presentation_list(self, sources, dst_folder):
        """create a list of presentation objects"""
        self.log_debug("Getting list of presentation objects.")
        presentation_list = []
        for src in sources:
            try:
                try:
                    pres_object = PresentationObject(dst_folder, **src)
                except TypeError:
                    pres_object = PresentationObject(dst_folder, src)
                self.log_debug(
                    f"Loading {pres_object.loader.diag_type} diagnostic from {pres_object.loader.path}."
                )
                presentation_list.append(pres_object.create_dict())
            except PresentationException as msg:
                self.log_warning(f"Can not present diagnostic: {msg}")
        return presentation_list

    def get_template(self, context, template):
        """get Jinja2 template file"""
        search_path = [".", "templates"]
        if "_se_cmd_cwd" in context:
            search_path.extend(
                [
                    context["_se_cmd_cwd"],
                    Path(context["_se_cmd_cwd"]) / "templates",
                ]
            )
        self.log_debug(f"Search path for template: {search_path}")

        loader = jinja2.FileSystemLoader(search_path)
        environment = jinja2.Environment(loader=loader)
        for name, function in j2filters().items():
            environment.filters[name] = function
        return environment.get_template(template)

    def get_project_and_issue(self, key, issue_subject):
        """Connect to Gitlab server, find and return correct issue and project based on user input"""

        gl = gitlab.Gitlab(SERVER_URL, private_token=key)
        try:
            project = gl.projects.get(PROJECT_ID)
        except (
            gitlab.GitlabAuthenticationError,
            requests.exceptions.ConnectionError,
        ) as e:
            msg = f"Could not log in to Gitlab server ({e})"
            self.log_error(msg)
            raise ScriptEngineTaskRunError()

        # Find issue or create if none exists
        issues = project.issues.list()
        issue = next((issue for issue in issues if issue.title == issue_subject), None)
        if issue is None:
            issue = project.issues.create({"title": issue_subject})
            issue.save()
        return project, issue


def create_anchor(title: str) -> str:
    """
    create correct anchors for Gitlab issue URLs
    """
    anchor = re.sub(r"[^a-zA-Z\s-]", "", title)
    anchor = re.sub(r"\s+(\-+\s*)?", "-", anchor)
    anchor = anchor.lower()
    return anchor

1	"""Presentation Task that uploads Data and Plots to the EC-Earth 4 Gitlab."""
2
3	import re	1✔
4	import urllib.parse	1✔
5	from pathlib import Path	1✔
6
7	import gitlab	1✔
8	import jinja2	1✔
9	import requests.exceptions	1✔
10	from scriptengine.exceptions import ScriptEngineTaskRunError	1✔
11	from scriptengine.jinja import filters as j2filters	1✔
12	from scriptengine.tasks.core import Task, timed_runner	1✔
13
14	from helpers.exceptions import PresentationException	1✔
15	from helpers.presentation_objects import PresentationObject	1✔
16
17	# Repository where monitoring results are posted
18	SERVER_URL = "https://git.smhi.se"	1✔
19	PROJECT_ID = 1982	1✔
20
21
22	class Gitlab(Task):	1✔
23	"""Gitlab Presentation Task"""
24
25	_required_arguments = (	1✔
26	"src",
27	"local_dst",
28	"subject",
29	"template",
30	"api_key",
31	)
32
33	def __init__(self, arguments=None):	1✔
34	Gitlab.check_arguments(arguments)	1✔
35	super().__init__(arguments)	1✔
36
37	@timed_runner	1✔
38	def run(self, context):	1✔
39	sources = self.getarg("src", context)	1✔
40	dst_folder = self.getarg("local_dst", context)	1✔
41	issue_subject = self.getarg("subject", context, parse_yaml=False)	1✔
42	template_path = self.getarg("template", context)	1✔
43	key = self.getarg("api_key", context)	1✔
44	self.log_info(f"Create Gitlab issue '{issue_subject}'.")	1✔
45	self.log_debug(f"Template: {template_path}, Source File(s): {sources}")	1✔
46
47	presentation_list = self.get_presentation_list(sources, dst_folder)	1✔
48	gitlab_template = self.get_template(context, template_path)	1✔
49	gitlab_template.globals["urlencode"] = urllib.parse.quote	1✔
50
51	self.log_debug("Connecting to Gitlab.")	1✔
52	project, issue = self.get_project_and_issue(key, issue_subject)	1✔
53
54	self.log_debug("Uploading attachments.")	1✔
55	issue.uploads = []	1✔
56	for item in presentation_list:	1✔
NEW 57	if item["presentation_type"] == "image":	×
NEW 58	file_name = Path(item["path"]).name	×
NEW 59	upload = project.upload(file_name, filepath=f"{dst_folder}/{file_name}")	×
NEW 60	item["image_link"] = upload[	×
61	"markdown"
62	] # string of the form "![]()" generated by Gitlab
63
64	# render the template and add as description
65	self.log_debug("Updating the issue description.")	1✔
66	issue.description = gitlab_template.render(	1✔
67	presentation_list=presentation_list,
68	issue_url=issue.web_url,
69	create_anchor=create_anchor,
70	)
71	self.log_debug("Saving issue.")	1✔
72	issue.save()	1✔
73
74	def get_presentation_list(self, sources, dst_folder):	1✔
75	"""create a list of presentation objects"""
76	self.log_debug("Getting list of presentation objects.")	1✔
77	presentation_list = []	1✔
78	for src in sources:	1✔
79	try:	1✔
80	try:	1✔
81	pres_object = PresentationObject(dst_folder, **src)	1✔
82	except TypeError:	1✔
83	pres_object = PresentationObject(dst_folder, src)	1✔
84	self.log_debug(	1✔
85	f"Loading {pres_object.loader.diag_type} diagnostic from {pres_object.loader.path}."
86	)
87	presentation_list.append(pres_object.create_dict())	1✔
88	except PresentationException as msg:	1✔
89	self.log_warning(f"Can not present diagnostic: {msg}")	1✔
90	return presentation_list	1✔
91
92	def get_template(self, context, template):	1✔
93	"""get Jinja2 template file"""
94	search_path = [".", "templates"]	1✔
95	if "_se_cmd_cwd" in context:	1✔
NEW 96	search_path.extend(	×
97	[
98	context["_se_cmd_cwd"],
99	Path(context["_se_cmd_cwd"]) / "templates",
100	]
101	)
102	self.log_debug(f"Search path for template: {search_path}")	1✔
103
104	loader = jinja2.FileSystemLoader(search_path)	1✔
105	environment = jinja2.Environment(loader=loader)	1✔
106	for name, function in j2filters().items():	1✔
107	environment.filters[name] = function	1✔
108	return environment.get_template(template)	1✔
109
110	def get_project_and_issue(self, key, issue_subject):	1✔
111	"""Connect to Gitlab server, find and return correct issue and project based on user input"""
112
113	gl = gitlab.Gitlab(SERVER_URL, private_token=key)	1✔
114	try:	1✔
115	project = gl.projects.get(PROJECT_ID)	1✔
116	except (	1✔
117	gitlab.GitlabAuthenticationError,
118	requests.exceptions.ConnectionError,
119	) as e:
120	msg = f"Could not log in to Gitlab server ({e})"	1✔
121	self.log_error(msg)	1✔
122	raise ScriptEngineTaskRunError()	1✔
123
124	# Find issue or create if none exists
NEW 125	issues = project.issues.list()	×
NEW 126	issue = next((issue for issue in issues if issue.title == issue_subject), None)	×
NEW 127	if issue is None:	×
NEW 128	issue = project.issues.create({"title": issue_subject})	×
NEW 129	issue.save()	×
NEW 130	return project, issue	×
131
132
133	def create_anchor(title: str) -> str:	1✔
134	"""
135	create correct anchors for Gitlab issue URLs
136	"""
137	anchor = re.sub(r"[^a-zA-Z\s-]", "", title)	1✔
138	anchor = re.sub(r"\s+(\-+\s*)?", "-", anchor)	1✔
139	anchor = anchor.lower()	1✔
140	return anchor	1✔

uwefladrich / scriptengine-tasks-ecearth / 14861206140

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous