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