Module ogr.services.gitlab.service

Expand source code
# Copyright Contributors to the Packit project.
# SPDX-License-Identifier: MIT

import logging
from typing import Optional

import gitlab

from ogr.abstract import GitUser
from ogr.exceptions import GitlabAPIException, OperationNotSupported
from ogr.factory import use_for_service
from ogr.services.base import BaseGitService, GitProject
from ogr.services.gitlab.project import GitlabProject
from ogr.services.gitlab.user import GitlabUser

logger = logging.getLogger(__name__)


@use_for_service("gitlab")  # anything containing a gitlab word in hostname
# + list of community-hosted instances based on the following list
# https://wiki.p2pfoundation.net/List_of_Community-Hosted_GitLab_Instances
@use_for_service("salsa.debian.org")
@use_for_service("git.fosscommunity.in")
@use_for_service("framagit.org")
@use_for_service("dev.gajim.org")
@use_for_service("git.coop")
@use_for_service("lab.libreho.st")
@use_for_service("git.linux-kernel.at")
@use_for_service("git.pleroma.social")
@use_for_service("git.silence.dev")
@use_for_service("code.videolan.org")
@use_for_service("source.puri.sm")
class GitlabService(BaseGitService):
    name = "gitlab"

    def __init__(self, token=None, instance_url=None, ssl_verify=True, **kwargs):
        super().__init__(token=token)
        self.instance_url = instance_url or "https://gitlab.com"
        self.token = token
        self.ssl_verify = ssl_verify
        self._gitlab_instance = None

        if kwargs:
            logger.warning(f"Ignored keyword arguments: {kwargs}")

    @property
    def gitlab_instance(self) -> gitlab.Gitlab:
        if not self._gitlab_instance:
            self._gitlab_instance = gitlab.Gitlab(
                url=self.instance_url,
                private_token=self.token,
                ssl_verify=self.ssl_verify,
            )
            if self.token:
                self._gitlab_instance.auth()
        return self._gitlab_instance

    @property
    def user(self) -> GitUser:
        return GitlabUser(service=self)

    def __str__(self) -> str:
        token_str = (
            f", token='{self.token[:1]}***{self.token[-1:]}'" if self.token else ""
        )
        ssl_str = ", ssl_verify=False" if not self.ssl_verify else ""
        return (
            f"GitlabService(instance_url='{self.instance_url}'"
            f"{token_str}"
            f"{ssl_str})"
        )

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, GitlabService):
            return False

        return (
            self.token == o.token  # type: ignore
            and self.instance_url == o.instance_url  # type: ignore
            and self.ssl_verify == o.ssl_verify  # type: ignore
        )

    def __hash__(self) -> int:
        return hash(str(self))

    def get_project(
        self,
        repo=None,
        namespace=None,
        is_fork=False,
        **kwargs,
    ) -> "GitlabProject":
        if is_fork:
            namespace = self.user.get_username()
        return GitlabProject(repo=repo, namespace=namespace, service=self, **kwargs)

    def get_project_from_project_id(self, iid: int) -> "GitlabProject":
        gitlab_repo = self.gitlab_instance.projects.get(iid)
        return GitlabProject(
            repo=gitlab_repo.attributes["path"],
            namespace=gitlab_repo.attributes["namespace"]["full_path"],
            service=self,
            gitlab_repo=gitlab_repo,
        )

    def change_token(self, new_token: str) -> None:
        self.token = new_token
        self._gitlab_instance = None

    def project_create(
        self,
        repo: str,
        namespace: Optional[str] = None,
        description: Optional[str] = None,
    ) -> "GitlabProject":
        data = {"name": repo}
        if namespace:
            try:
                group = self.gitlab_instance.groups.get(namespace)
            except gitlab.GitlabGetError as ex:
                raise GitlabAPIException(f"Group {namespace} not found.") from ex
            data["namespace_id"] = group.id

        if description:
            data["description"] = description
        try:
            new_project = self.gitlab_instance.projects.create(data)
        except gitlab.GitlabCreateError as ex:
            raise GitlabAPIException("Project already exists") from ex
        return GitlabProject(
            repo=repo,
            namespace=namespace,
            service=self,
            gitlab_repo=new_project,
        )

    def list_projects(
        self,
        namespace: Optional[str] = None,
        user: Optional[str] = None,
        search_pattern: Optional[str] = None,
        language: Optional[str] = None,
    ) -> list[GitProject]:
        if namespace:
            group = self.gitlab_instance.groups.get(namespace)
            projects = group.projects.list(all=True)
        elif user:
            user_object = self.gitlab_instance.users.list(username=user)[0]
            projects = user_object.projects.list(all=True)
        else:
            raise OperationNotSupported

        if language:
            # group.projects.list gives us a GroupProject instance
            # in order to be able to filter by language we need Project instance
            projects_to_convert = [
                self.gitlab_instance.projects.get(item.attributes["id"])
                for item in projects
                if language
                in self.gitlab_instance.projects.get(item.attributes["id"]).languages()
            ]
        else:
            projects_to_convert = projects
        return [
            GitlabProject(
                repo=project.attributes["path"],
                namespace=project.attributes["namespace"]["full_path"],
                service=self,
            )
            for project in projects_to_convert
        ]

Classes

class GitlabService (token=None, instance_url=None, ssl_verify=True, **kwargs)

Attributes

instance_url : str
URL of the git forge instance.
Expand source code
@use_for_service("gitlab")  # anything containing a gitlab word in hostname
# + list of community-hosted instances based on the following list
# https://wiki.p2pfoundation.net/List_of_Community-Hosted_GitLab_Instances
@use_for_service("salsa.debian.org")
@use_for_service("git.fosscommunity.in")
@use_for_service("framagit.org")
@use_for_service("dev.gajim.org")
@use_for_service("git.coop")
@use_for_service("lab.libreho.st")
@use_for_service("git.linux-kernel.at")
@use_for_service("git.pleroma.social")
@use_for_service("git.silence.dev")
@use_for_service("code.videolan.org")
@use_for_service("source.puri.sm")
class GitlabService(BaseGitService):
    name = "gitlab"

    def __init__(self, token=None, instance_url=None, ssl_verify=True, **kwargs):
        super().__init__(token=token)
        self.instance_url = instance_url or "https://gitlab.com"
        self.token = token
        self.ssl_verify = ssl_verify
        self._gitlab_instance = None

        if kwargs:
            logger.warning(f"Ignored keyword arguments: {kwargs}")

    @property
    def gitlab_instance(self) -> gitlab.Gitlab:
        if not self._gitlab_instance:
            self._gitlab_instance = gitlab.Gitlab(
                url=self.instance_url,
                private_token=self.token,
                ssl_verify=self.ssl_verify,
            )
            if self.token:
                self._gitlab_instance.auth()
        return self._gitlab_instance

    @property
    def user(self) -> GitUser:
        return GitlabUser(service=self)

    def __str__(self) -> str:
        token_str = (
            f", token='{self.token[:1]}***{self.token[-1:]}'" if self.token else ""
        )
        ssl_str = ", ssl_verify=False" if not self.ssl_verify else ""
        return (
            f"GitlabService(instance_url='{self.instance_url}'"
            f"{token_str}"
            f"{ssl_str})"
        )

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, GitlabService):
            return False

        return (
            self.token == o.token  # type: ignore
            and self.instance_url == o.instance_url  # type: ignore
            and self.ssl_verify == o.ssl_verify  # type: ignore
        )

    def __hash__(self) -> int:
        return hash(str(self))

    def get_project(
        self,
        repo=None,
        namespace=None,
        is_fork=False,
        **kwargs,
    ) -> "GitlabProject":
        if is_fork:
            namespace = self.user.get_username()
        return GitlabProject(repo=repo, namespace=namespace, service=self, **kwargs)

    def get_project_from_project_id(self, iid: int) -> "GitlabProject":
        gitlab_repo = self.gitlab_instance.projects.get(iid)
        return GitlabProject(
            repo=gitlab_repo.attributes["path"],
            namespace=gitlab_repo.attributes["namespace"]["full_path"],
            service=self,
            gitlab_repo=gitlab_repo,
        )

    def change_token(self, new_token: str) -> None:
        self.token = new_token
        self._gitlab_instance = None

    def project_create(
        self,
        repo: str,
        namespace: Optional[str] = None,
        description: Optional[str] = None,
    ) -> "GitlabProject":
        data = {"name": repo}
        if namespace:
            try:
                group = self.gitlab_instance.groups.get(namespace)
            except gitlab.GitlabGetError as ex:
                raise GitlabAPIException(f"Group {namespace} not found.") from ex
            data["namespace_id"] = group.id

        if description:
            data["description"] = description
        try:
            new_project = self.gitlab_instance.projects.create(data)
        except gitlab.GitlabCreateError as ex:
            raise GitlabAPIException("Project already exists") from ex
        return GitlabProject(
            repo=repo,
            namespace=namespace,
            service=self,
            gitlab_repo=new_project,
        )

    def list_projects(
        self,
        namespace: Optional[str] = None,
        user: Optional[str] = None,
        search_pattern: Optional[str] = None,
        language: Optional[str] = None,
    ) -> list[GitProject]:
        if namespace:
            group = self.gitlab_instance.groups.get(namespace)
            projects = group.projects.list(all=True)
        elif user:
            user_object = self.gitlab_instance.users.list(username=user)[0]
            projects = user_object.projects.list(all=True)
        else:
            raise OperationNotSupported

        if language:
            # group.projects.list gives us a GroupProject instance
            # in order to be able to filter by language we need Project instance
            projects_to_convert = [
                self.gitlab_instance.projects.get(item.attributes["id"])
                for item in projects
                if language
                in self.gitlab_instance.projects.get(item.attributes["id"]).languages()
            ]
        else:
            projects_to_convert = projects
        return [
            GitlabProject(
                repo=project.attributes["path"],
                namespace=project.attributes["namespace"]["full_path"],
                service=self,
            )
            for project in projects_to_convert
        ]

Ancestors

Class variables

var name

Instance variables

var gitlab_instance : gitlab.client.Gitlab
Expand source code
@property
def gitlab_instance(self) -> gitlab.Gitlab:
    if not self._gitlab_instance:
        self._gitlab_instance = gitlab.Gitlab(
            url=self.instance_url,
            private_token=self.token,
            ssl_verify=self.ssl_verify,
        )
        if self.token:
            self._gitlab_instance.auth()
    return self._gitlab_instance

Methods

def get_project_from_project_id(self, iid: int) ‑> GitlabProject
Expand source code
def get_project_from_project_id(self, iid: int) -> "GitlabProject":
    gitlab_repo = self.gitlab_instance.projects.get(iid)
    return GitlabProject(
        repo=gitlab_repo.attributes["path"],
        namespace=gitlab_repo.attributes["namespace"]["full_path"],
        service=self,
        gitlab_repo=gitlab_repo,
    )

Inherited members