Source code for semantic_release.hvcs.gitlab

"""Helper code for interacting with a Gitlab remote VCS"""

from __future__ import annotations

import logging
import os
from functools import lru_cache
from pathlib import PurePosixPath
from re import compile as regexp
from typing import TYPE_CHECKING

import gitlab
import gitlab.exceptions
import gitlab.v4
import gitlab.v4.objects
from urllib3.util.url import Url, parse_url

from semantic_release.cli.util import noop_report
from semantic_release.errors import UnexpectedResponse
from semantic_release.helpers import logged_function
from semantic_release.hvcs.remote_hvcs_base import RemoteHvcsBase
from semantic_release.hvcs.util import suppress_not_found

if TYPE_CHECKING:
    from typing import Any, Callable

    from gitlab.v4.objects import Project as GitLabProject


log = logging.getLogger(__name__)


# Globals
log = logging.getLogger(__name__)


[docs] class Gitlab(RemoteHvcsBase): """Gitlab HVCS interface for interacting with Gitlab repositories""" DEFAULT_ENV_TOKEN_NAME = "GITLAB_TOKEN" # noqa: S105 # purposefully not CI_JOB_TOKEN as it is not a personal access token, # It is missing the permission to push to the repository, but has all others (releases, packages, etc.) DEFAULT_DOMAIN = "gitlab.com" def __init__( self, remote_url: str, *, hvcs_domain: str | None = None, token: str | None = None, allow_insecure: bool = False, **_kwargs: Any, ) -> None: super().__init__(remote_url) self.token = token self.project_namespace = f"{self.owner}/{self.repo_name}" self._project: GitLabProject | None = None domain_url = self._normalize_url( hvcs_domain or os.getenv("CI_SERVER_URL", "") or f"https://{self.DEFAULT_DOMAIN}", allow_insecure=allow_insecure, ) # Strip any auth, query or fragment from the domain self._hvcs_domain = parse_url( Url( scheme=domain_url.scheme, host=domain_url.host, port=domain_url.port, path=str(PurePosixPath(domain_url.path or "/")), ).url.rstrip("/") ) self._client = gitlab.Gitlab(self.hvcs_domain.url, private_token=self.token) self._api_url = parse_url(self._client.api_url) @property def project(self) -> GitLabProject: if self._project is None: self._project = self._client.projects.get(self.project_namespace) return self._project @lru_cache(maxsize=1) def _get_repository_owner_and_name(self) -> tuple[str, str]: """ Get the repository owner and name from GitLab CI environment variables, if available, otherwise from parsing the remote url """ if "CI_PROJECT_NAMESPACE" in os.environ and "CI_PROJECT_NAME" in os.environ: log.debug("getting repository owner and name from environment variables") return os.environ["CI_PROJECT_NAMESPACE"], os.environ["CI_PROJECT_NAME"] return super()._get_repository_owner_and_name()
[docs] @logged_function(log) def create_release( self, tag: str, release_notes: str, prerelease: bool = False, # noqa: ARG002 assets: list[str] | None = None, # noqa: ARG002 noop: bool = False, ) -> str: """ Create a release in a remote VCS, adding any release notes and assets to it :param tag: The tag to create the release for :param release_notes: The changelog description for this version only :param prerelease: This parameter has no effect in GitLab :param assets: A list of paths to files to upload as assets (TODO: not implemented) :param noop: If True, do not perform any actions, only log intents :return: The tag of the release :raises: GitlabAuthenticationError: If authentication is not correct :raises: GitlabCreateError: If the server cannot perform the request """ if noop: noop_report(f"would have created a release for tag {tag}") return tag log.info("Creating release for %s", tag) # ref: https://docs.gitlab.com/ee/api/releases/index.html#create-a-release self.project.releases.create( { "name": tag, "tag_name": tag, "tag_message": tag, "description": release_notes, } ) log.info("Successfully created release for %s", tag) return tag
[docs] @logged_function(log) @suppress_not_found def get_release_by_tag(self, tag: str) -> gitlab.v4.objects.ProjectRelease | None: """ Get a release by its tag name. :param tag: The tag name to get the release for :return: gitlab.v4.objects.ProjectRelease or None if not found :raises: gitlab.exceptions.GitlabAuthenticationError: If the user is not authenticated """ try: return self.project.releases.get(tag) except gitlab.exceptions.GitlabGetError: log.debug("Release %s not found", tag) return None except KeyError as err: raise UnexpectedResponse("JSON response is missing commit.id") from err
[docs] @logged_function(log) def edit_release_notes( # type: ignore[override] self, release: gitlab.v4.objects.ProjectRelease, release_notes: str, ) -> str: """ Update the release notes for a given release. :param release: The release object to update :param release_notes: The new release notes :return: The release id :raises: GitlabAuthenticationError: If authentication is not correct :raises: GitlabUpdateError: If the server cannot perform the request """ log.info( "Updating release %s [%s]", release.name, release.attributes.get("commit", {}).get("id"), ) release.description = release_notes release.save() return str(release.get_id())
[docs] @logged_function(log) def create_or_update_release( self, tag: str, release_notes: str, prerelease: bool = False ) -> str: """ Create or update a release for the given tag in a remote VCS. :param tag: The tag to create or update the release for :param release_notes: The changelog description for this version only :param prerelease: This parameter has no effect in GitLab :return: The release id :raises ValueError: If the release could not be created or updated :raises gitlab.exceptions.GitlabAuthenticationError: If the user is not authenticated :raises GitlabUpdateError: If the server cannot perform the request """ try: return self.create_release( tag=tag, release_notes=release_notes, prerelease=prerelease ) except gitlab.GitlabCreateError: log.info( "New release %s could not be created for project %s", tag, self.project_namespace, ) if (release_obj := self.get_release_by_tag(tag)) is None: raise ValueError( f"release for tag {tag} could not be found, and could not be created" ) log.debug( "Found existing release commit %s, updating", release_obj.commit.get("id") ) # If this errors we let it die return self.edit_release_notes( release=release_obj, release_notes=release_notes, )
[docs] def remote_url(self, use_token: bool = True) -> str: """Get the remote url including the token for authentication if requested""" if not (self.token and use_token): return self._remote_url return self.create_server_url( auth=f"gitlab-ci-token:{self.token}", path=f"{self.project_namespace}.git", )
[docs] def compare_url(self, from_rev: str, to_rev: str) -> str: return self.create_repo_url(repo_path=f"/-/compare/{from_rev}...{to_rev}")
[docs] def commit_hash_url(self, commit_hash: str) -> str: return self.create_repo_url(repo_path=f"/-/commit/{commit_hash}")
[docs] def issue_url(self, issue_num: str | int) -> str: # Strips off any character prefix like '#' that usually exists if isinstance(issue_num, str) and ( match := regexp(r"(\d+)$").search(issue_num) ): try: issue_num = int(match.group(1)) except ValueError: return "" if isinstance(issue_num, int): return self.create_repo_url(repo_path=f"/-/issues/{issue_num}") return ""
[docs] def merge_request_url(self, mr_number: str | int) -> str: # Strips off any character prefix like '!' that usually exists if isinstance(mr_number, str) and ( match := regexp(r"(\d+)$").search(mr_number) ): try: mr_number = int(match.group(1)) except ValueError: return "" if isinstance(mr_number, int): return self.create_repo_url(repo_path=f"/-/merge_requests/{mr_number}") return ""
[docs] def pull_request_url(self, pr_number: str | int) -> str: return self.merge_request_url(mr_number=pr_number)
[docs] def upload_dists(self, tag: str, dist_glob: str) -> int: return super().upload_dists(tag, dist_glob)
[docs] def get_changelog_context_filters(self) -> tuple[Callable[..., Any], ...]: return ( self.create_server_url, self.create_repo_url, self.commit_hash_url, self.compare_url, self.issue_url, self.merge_request_url, self.pull_request_url, )
RemoteHvcsBase.register(Gitlab)