#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @Author: Jialiang Shi
import logging
from base64 import b64decode
from urllib.parse import quote_plus, unquote_plus
from typing import Any, Dict, List, Optional
import requests
from gerrit import GerritClient
from gerrit.utils.common import params_creator
from gerrit.utils.gerritbase import GerritBase
from gerrit.utils.exceptions import (
BranchNotFoundError,
BranchAlreadyExistsError,
ConflictError,
GerritAPIException,
)
logger = logging.getLogger(__name__)
[docs]
class GerritProjectBranch(GerritBase):
def __init__(self, name: str, project: str, gerrit: GerritClient) -> None:
self.name = name
self.project = project
self.gerrit = gerrit
self.endpoint = f"/projects/{self.project}/branches/{quote_plus(self.name)}"
super().__init__()
def __str__(self) -> str:
return self.name
[docs]
def get_file_content(self, file: str, decode: bool = False) -> Any:
"""
Gets the content of a file from the HEAD revision of a certain branch.
The content is returned as base64 encoded string.
:param file: the file path
:param decode: Decode bas64 to plain text.
:return:
"""
result = self.gerrit.get(self.endpoint + f"/files/{quote_plus(file)}/content")
if decode:
return b64decode(result).decode("utf-8")
return result
[docs]
def is_mergeable(self, input_: Dict[str, Any]) -> Any:
"""
Gets whether the source is mergeable with the target branch.
.. code-block:: python
input_ = {
'source': 'testbranch',
'strategy': 'recursive'
}
result = stable.is_mergeable(input_)
:param input_: the MergeInput entity,
https://gerrit-review.googlesource.com/Documentation/rest-api-changes.html#merge-input
:return:
"""
source = input_.get("source")
try:
project = self.gerrit.projects.get(unquote_plus(self.project))
project.branches.get(name=source)
except requests.exceptions.HTTPError as error:
if error.response.status_code == 404:
message = f"Source Branch {source} does not exist"
logger.error(message)
raise BranchNotFoundError(message)
raise GerritAPIException(str(error)) from error
return self.gerrit.get(self.endpoint + "/mergeable", params=input_)
[docs]
def get_reflog(self) -> Any:
"""
Gets the reflog of a certain branch.
:return:
"""
return self.gerrit.get(self.endpoint + "/reflog")
[docs]
def delete(self) -> None:
"""
Delete a branch.
:return:
"""
self.gerrit.delete(self.endpoint)
[docs]
class GerritProjectBranches:
branch_prefix = "refs/heads/"
def __init__(self, project: str, gerrit: GerritClient) -> None:
self.project = project
self.gerrit = gerrit
self.endpoint = f"/projects/{self.project}/branches"
[docs]
def list(self, pattern_dispatcher: Optional[Dict[str, Any]] = None, limit: int = 25, skip: int = 0) -> List[Any]:
"""
List the branches of a project.
:param pattern_dispatcher: Dict of pattern type with respective
pattern value: {('match'|'regex') : value}
:param limit: Limit the number of branches to be included in the results.
:param skip: Skip the given number of branches from the beginning of the list.
:return:
"""
params = params_creator(
(("n", limit), ("s", skip)),
{"match": "m", "regex": "r"},
pattern_dispatcher,
)
return self.gerrit.get(self.endpoint + "/", params=params)
[docs]
def get(self, name: str) -> Any:
"""
get a branch by ref
:param name: branch ref name
:return:
"""
try:
result = self.gerrit.get(self.endpoint + f"/{quote_plus(name)}")
ref = result.get("ref")
name = ref.replace(self.branch_prefix, "")
return GerritProjectBranch(
name=name, project=self.project, gerrit=self.gerrit
)
except requests.exceptions.HTTPError as error:
if error.response.status_code == 404:
message = f"Branch {name} does not exist"
raise BranchNotFoundError(message)
raise GerritAPIException from error
[docs]
def create(self, name: str, input_: Dict[str, Any]) -> Any:
"""
Creates a new branch.
.. code-block:: python
input_ = {
'revision': '76016386a0d8ecc7b6be212424978bb45959d668'
}
project = client.projects.get('myproject')
new_branch = project.branches.create('stable', input_)
:param name: the branch name
:param input_: the BranchInput entity,
https://gerrit-review.googlesource.com/Documentation/rest-api-projects.html#branch-info
:return:
"""
try:
self.gerrit.put(
self.endpoint + f"/{quote_plus(name)}",
json=input_,
headers=self.gerrit.default_headers,
)
except ConflictError:
message = f"Branch {name} already exists"
logger.error(message)
raise BranchAlreadyExistsError(message)
return self.get(name)
[docs]
def delete(self, name: str) -> None:
"""
Delete a branch.
:param name: branch ref name
:return:
"""
self.get(name)
self.gerrit.delete(self.endpoint + f"/{quote_plus(name)}")
[docs]
def delete_branches(self, input_: Dict[str, Any]) -> None:
"""
Delete one or more branches.
.. code-block:: python
input_ = {
"branches": [
"stable-1.0",
"stable-2.0"
]
}
project = client.projects.get('myproject')
project.branches.delete_branches(input_)
:param input_: the DeleteBranchesInput entity,
https://gerrit-review.googlesource.com/Documentation/rest-api-projects.html#delete-branches-input
:return:
"""
self.gerrit.post(
self.endpoint + ":delete",
json=input_,
headers=self.gerrit.default_headers,
)