"""Module for interfacing the ESS Gitlab wrapper repositories."""
import functools
import logging
from typing import Dict, List, Sequence, Collection, Optional
import gitlab
from gitlab.exceptions import GitlabGetError, GitlabCreateError
from gitlab.v4.objects import Project, Group, GroupProject, ProjectCommit, ProjectTag
from e3_build_tools import utils
from e3_build_tools.exceptions import (
NoModuleChangesException,
MissingReferenceException,
FileNotFoundException,
)
[docs]
ESS_GITLAB_URL = "https://gitlab.esss.lu.se"
[docs]
GITLAB_ID_E3_GROUP = 215
[docs]
logger = logging.getLogger(__name__)
[docs]
GitLabCommitPayload = Dict[str, Sequence[Collection[str]]]
[docs]
class GitLabInterface:
"""Interface to GitLab."""
def __init__(
self,
address: str = ESS_GITLAB_URL,
private_token: Optional[str] = None,
config: Optional[str] = None,
top_group: int = GITLAB_ID_E3_GROUP,
) -> None:
"""Initialise the GitLabInterface object."""
[docs]
self.top_group = top_group
if private_token is not None:
self._gl = gitlab.Gitlab(self.address, private_token=private_token)
elif config is not None:
self._gl = gitlab.Gitlab.from_config(config)
else:
self._gl = gitlab.Gitlab(self.address)
[docs]
def get_all_projects(self) -> List[GroupProject]:
"""Returns a list of all GroupProjects in the e3 space.
The object returned by `Gitlab.groups.get().projects.list()` is `GroupProject`.
This needs to be converted to `Project` to have access to all attributes, but
we do not convert at this stage (to keep down the number of calls).
"""
logger.debug("Retrieving project gitlab metadata")
projects = self._gl.groups.get(self.top_group).projects.list(
all=True, include_subgroups=True
)
logger.debug(f"Loading projects from group {self.top_group}: {projects}")
return projects
@functools.lru_cache(maxsize=None)
[docs]
def get_project_by_id(self, _id: int) -> Project:
"""Return a GitLab `Project` object based on the ID.
Caches previous results to reduce the amount of API calls.
We are using the Least Recently Used (LRU) cache but with no limit to the
number of elements stored, effectively turning it into an unbound cache.
"""
return self._gl.projects.get(_id)
[docs]
def get_current_user_id(self) -> int:
"""Return the current user ID.
Raises:
GitlabAuthenticationError: If authentication is not correct.
"""
self._gl.auth()
return self._gl.user.id
[docs]
def create_new_project(self, name: str, group_id: int) -> Project:
"""Create a new GitLab project.
Raises:
GitlabCreateError: If project already exists in the group.
"""
try:
project = self._gl.projects.create(
{"name": name, "namespace_id": group_id, "visibility": "public"}
)
return project
except GitlabCreateError:
group_name = self.get_group_by_id(group_id).full_path
logger.warning("The project %s already exists in %s.", name, group_name)
raise
[docs]
def get_group_by_id(self, _id: int) -> Group:
"""Return a GitLab `Group` object based on the ID.
Caches previous results to reduce the amount of API calls.
We are using the Least Recently Used (LRU) cache but with no limit to the
number of elements stored, effectively turning it into an unbound cache.
"""
return self._gl.groups.get(_id)
[docs]
class WrapperRegistry(GitLabInterface):
"""Registry for wrappers and interface to GitLab.
The class stores information about the wrapper repositories, and implements
the required interfaces to the ESS GitLab repositories.
"""
def __init__(
self,
**kwargs,
) -> None:
"""Initialise the WrapperRegistry object."""
super().__init__(**kwargs)
self._projects = self.get_all_wrapper_projects()
[docs]
def get_all_wrapper_projects(self) -> Dict[str, int]:
"""Return a dictionary with the project id for each project.
The returned data is a dictionary with the project name as the keys, and
the internal GitLab project ID as the data.
"""
logger.debug("Finding wrappers")
projects = {
project.path.lower(): project.id
for project in self.get_all_projects()
if project.namespace["name"] == "wrappers"
}
return projects
[docs]
def get_project(self, name: str) -> Project:
"""Return a GitLab `Project` object based on the name.
Raises:
KeyError: If module name not found.
"""
return self.get_project_by_id(self.get_project_id(name))
[docs]
def get_project_id(self, name: str) -> int:
"""Return the GitLab ID for a project name.
Raises:
KeyError: If module name not found.
"""
qualified_name = f"e3-{name}"
try:
project = self._projects[qualified_name]
except KeyError:
logger.warning("Project %s not found.", qualified_name)
raise
return project
@functools.lru_cache(maxsize=None)
[docs]
def get_config_from(self, name: str, ref: str, *, file: str) -> str:
"""Return a module revision's configuration.
The function fetches the configuration for a wrapper from the GitLab
repository at a specified reference (e.g., commit hash, tag, branch).
Raises:
MissingReferenceException: If the reference does not exists on the repository.
FileNotFoundException: If `name` does not exist or if `file` does not exist for the given reference.
"""
try:
return (
self.get_project(name)
.files.raw(file_path=file, ref=ref)
.decode("utf-8")
)
except GitlabGetError as e:
if "404 Commit Not Found" in str(e):
raise MissingReferenceException from e
elif "404 File Not Found" in str(e):
raise FileNotFoundException from e
else:
raise e
[docs]
def read_version_from(
self,
name: str,
ref: str,
*,
config: str,
var: str,
validate: bool = True,
) -> str:
"""Return the configured version for a version of a module.
Raises:
KeyError: If `name` is not a valid e3-module or `file` does not exist for the given reference.
MissingReferenceException: If the reference does not exists on the repository.
FileNotFoundException: If `name` does not exist or if `file` does not exist for the given reference.
"""
module_version = utils.read_makefile_definitions(
self.get_config_from(
name,
ref,
file=config,
),
)[var]
logger.debug(
f"Module version ({ref}) of {name} is read as '{module_version}' from {var!r} in {config!r}"
)
if validate:
module_version = utils.ensure_build_number(module_version)
return module_version
[docs]
def create_commit_payload(
self,
substitutions: Dict[str, Dict[str, str]],
starting_reference: str,
name: str,
branch: str,
) -> GitLabCommitPayload:
"""Return representation of JSON data to be used by `commit_change()`.
Raises:
NoModuleChangesException: If there are no changes to be applied to the module.
"""
logger.info(
f"Preparing commit to create {name} {substitutions['configure/CONFIG_MODULE']['E3_MODULE_VERSION']} ({starting_reference})"
)
changes = self._create_change_to_files(
name=name,
substitutions=substitutions,
starting_ref=starting_reference,
)
ret = {
"branch": branch,
"commit_message": "Update versions (auto-generated commit) [skip ci]",
"actions": changes,
}
branch_list = [x.name for x in self.get_project(name).branches.list(all=True)]
if starting_reference not in branch_list:
commit = self.get_project(name).commits.get(starting_reference)
# start_sha accepts only the full commit id hash
ret["start_sha"] = commit.id
else:
ret["start_branch"] = starting_reference
return ret
[docs]
def update_submodule(
self, name: str, submodule_path: str, sha: str, branch: str
) -> None:
"""Update the submodule reference."""
project = self.get_project(name)
logger.debug(f"Updating project submodule: path {submodule_path}, ref {sha}")
project.update_submodule(
submodule=submodule_path,
branch=branch,
commit_sha=sha,
commit_message="Update submodule [skip ci]",
)
[docs]
def commit_change(self, name: str, payload: GitLabCommitPayload) -> ProjectCommit:
"""Commit changes to project on gitlab."""
project = self.get_project(name)
logger.debug(
f"Creating commit for project {project.path} with payload {payload}"
)
commit = project.commits.create(payload)
logger.debug(f"Created commit {commit}")
return commit
[docs]
def tag_project(
self, name: str, new_tag: str, ref: Optional[str] = None
) -> ProjectTag:
"""Apply a tag to a wrapper repository.
Raises:
KeyError: If `name` is not a valid e3-module.
"""
project = self.get_project(name)
if ref is None:
ref = project.default_branch
logger.debug(f"Tagging project {project.path} with tag {new_tag} on ref {ref}")
return project.tags.create({"tag_name": new_tag, "ref": ref})
[docs]
def get_commit_hash(self, name: str, reference: str) -> str:
"""Return the commit hash for a specified reference.
Raises:
KeyError: If `name` is not a valid e3-module.
"""
project = self.get_project(name)
commits = project.commits.list(ref_name=reference)
latest_commit, *_ = commits
return latest_commit.id
def _create_change_to_files(
self,
name: str,
substitutions: Dict[str, Dict[str, str]],
starting_ref: str,
) -> List[Dict[str, str]]:
"""Create a changes for configuration file(s) in a wrapper repository.
Raises:
KeyError: If `name` is not a valid e3-module.
NoModuleChangeException: If there are no changes to the files.
MissingReferenceException: If reference is missing from the repository.
FileNotFoundException: If file don't exists on the repository.
"""
project = self.get_project(name)
changes = []
for file, change in substitutions.items():
old_fcontent = self.get_config_from(name, starting_ref, file=file)
new_fcontent = utils.modify_makefile_definitions(old_fcontent, change)
if new_fcontent and new_fcontent != old_fcontent:
changes.append(
{
"action": "update",
"file_path": file,
"content": new_fcontent,
}
)
if not changes:
logger.warn(
f"No changes to any file - skipping commit to {project.path} starting from ref {starting_ref} with {substitutions}"
)
commits = project.commits.list(ref_name=starting_ref)
commit, *_ = commits
logger.debug(f"Using the latest commit on {starting_ref}: {commit}")
raise NoModuleChangesException
logging.debug(f"Version {starting_ref!r} of {name!r} will have {changes}")
return changes