# -*- coding: utf-8 -*-
#
# Copyright (C) 2013-2017 Gauvain Pocentek
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see .
"""Wrapper for the GitLab API."""
import os
import time
from typing import Any, cast, Dict, List, Optional, Tuple, TYPE_CHECKING, Union
import requests
import requests.utils
from requests_toolbelt.multipart.encoder import MultipartEncoder # type: ignore
import gitlab.config
import gitlab.const
import gitlab.exceptions
from gitlab import utils
REDIRECT_MSG = (
"python-gitlab detected a {status_code} ({reason!r}) redirection. You must update "
"your GitLab URL to the correct URL to avoid issues. The redirection was from: "
"{source!r} to {target!r}"
)
class Gitlab(object):
"""Represents a GitLab server connection.
Args:
url: The URL of the GitLab server (defaults to https://gitlab.com).
private_token: The user private token
oauth_token: An oauth token
job_token: A CI job token
ssl_verify: Whether SSL certificates should be validated. If
the value is a string, it is the path to a CA file used for
certificate validation.
timeout: Timeout to use for requests to the GitLab server.
http_username: Username for HTTP authentication
http_password: Password for HTTP authentication
api_version: Gitlab API version to use (support for 4 only)
pagination: Can be set to 'keyset' to use keyset pagination
order_by: Set order_by globally
user_agent: A custom user agent to use for making HTTP requests.
retry_transient_errors: Whether to retry after 500, 502, 503, or
504 responses. Defaults to False.
"""
def __init__(
self,
url: Optional[str] = None,
private_token: Optional[str] = None,
oauth_token: Optional[str] = None,
job_token: Optional[str] = None,
ssl_verify: Union[bool, str] = True,
http_username: Optional[str] = None,
http_password: Optional[str] = None,
timeout: Optional[float] = None,
api_version: str = "4",
session: Optional[requests.Session] = None,
per_page: Optional[int] = None,
pagination: Optional[str] = None,
order_by: Optional[str] = None,
user_agent: str = gitlab.const.USER_AGENT,
retry_transient_errors: bool = False,
) -> None:
self._api_version = str(api_version)
self._server_version: Optional[str] = None
self._server_revision: Optional[str] = None
self._base_url = self._get_base_url(url)
self._url = f"{self._base_url}/api/v{api_version}"
#: Timeout to use for requests to gitlab server
self.timeout = timeout
self.retry_transient_errors = retry_transient_errors
#: Headers that will be used in request to GitLab
self.headers = {"User-Agent": user_agent}
#: Whether SSL certificates should be validated
self.ssl_verify = ssl_verify
self.private_token = private_token
self.http_username = http_username
self.http_password = http_password
self.oauth_token = oauth_token
self.job_token = job_token
self._set_auth_info()
#: Create a session object for requests
self.session = session or requests.Session()
self.per_page = per_page
self.pagination = pagination
self.order_by = order_by
# We only support v4 API at this time
if self._api_version not in ("4",):
raise ModuleNotFoundError(name=f"gitlab.v{self._api_version}.objects")
# NOTE: We must delay import of gitlab.v4.objects until now or
# otherwise it will cause circular import errors
import gitlab.v4.objects
objects = gitlab.v4.objects
self._objects = objects
self.broadcastmessages = objects.BroadcastMessageManager(self)
"""See :class:`~gitlab.v4.objects.BroadcastMessageManager`"""
self.deploykeys = objects.DeployKeyManager(self)
"""See :class:`~gitlab.v4.objects.DeployKeyManager`"""
self.deploytokens = objects.DeployTokenManager(self)
"""See :class:`~gitlab.v4.objects.DeployTokenManager`"""
self.geonodes = objects.GeoNodeManager(self)
"""See :class:`~gitlab.v4.objects.GeoNodeManager`"""
self.gitlabciymls = objects.GitlabciymlManager(self)
"""See :class:`~gitlab.v4.objects.GitlabciymlManager`"""
self.gitignores = objects.GitignoreManager(self)
"""See :class:`~gitlab.v4.objects.GitignoreManager`"""
self.groups = objects.GroupManager(self)
"""See :class:`~gitlab.v4.objects.GroupManager`"""
self.hooks = objects.HookManager(self)
"""See :class:`~gitlab.v4.objects.HookManager`"""
self.issues = objects.IssueManager(self)
"""See :class:`~gitlab.v4.objects.IssueManager`"""
self.issues_statistics = objects.IssuesStatisticsManager(self)
"""See :class:`~gitlab.v4.objects.IssuesStatisticsManager`"""
self.keys = objects.KeyManager(self)
"""See :class:`~gitlab.v4.objects.KeyManager`"""
self.ldapgroups = objects.LDAPGroupManager(self)
"""See :class:`~gitlab.v4.objects.LDAPGroupManager`"""
self.licenses = objects.LicenseManager(self)
"""See :class:`~gitlab.v4.objects.LicenseManager`"""
self.namespaces = objects.NamespaceManager(self)
"""See :class:`~gitlab.v4.objects.NamespaceManager`"""
self.mergerequests = objects.MergeRequestManager(self)
"""See :class:`~gitlab.v4.objects.MergeRequestManager`"""
self.notificationsettings = objects.NotificationSettingsManager(self)
"""See :class:`~gitlab.v4.objects.NotificationSettingsManager`"""
self.projects = objects.ProjectManager(self)
"""See :class:`~gitlab.v4.objects.ProjectManager`"""
self.runners = objects.RunnerManager(self)
"""See :class:`~gitlab.v4.objects.RunnerManager`"""
self.settings = objects.ApplicationSettingsManager(self)
"""See :class:`~gitlab.v4.objects.ApplicationSettingsManager`"""
self.appearance = objects.ApplicationAppearanceManager(self)
"""See :class:`~gitlab.v4.objects.ApplicationAppearanceManager`"""
self.sidekiq = objects.SidekiqManager(self)
"""See :class:`~gitlab.v4.objects.SidekiqManager`"""
self.snippets = objects.SnippetManager(self)
"""See :class:`~gitlab.v4.objects.SnippetManager`"""
self.users = objects.UserManager(self)
"""See :class:`~gitlab.v4.objects.UserManager`"""
self.todos = objects.TodoManager(self)
"""See :class:`~gitlab.v4.objects.TodoManager`"""
self.dockerfiles = objects.DockerfileManager(self)
"""See :class:`~gitlab.v4.objects.DockerfileManager`"""
self.events = objects.EventManager(self)
"""See :class:`~gitlab.v4.objects.EventManager`"""
self.audit_events = objects.AuditEventManager(self)
"""See :class:`~gitlab.v4.objects.AuditEventManager`"""
self.features = objects.FeatureManager(self)
"""See :class:`~gitlab.v4.objects.FeatureManager`"""
self.pagesdomains = objects.PagesDomainManager(self)
"""See :class:`~gitlab.v4.objects.PagesDomainManager`"""
self.user_activities = objects.UserActivitiesManager(self)
"""See :class:`~gitlab.v4.objects.UserActivitiesManager`"""
self.applications = objects.ApplicationManager(self)
"""See :class:`~gitlab.v4.objects.ApplicationManager`"""
self.variables = objects.VariableManager(self)
"""See :class:`~gitlab.v4.objects.VariableManager`"""
self.personal_access_tokens = objects.PersonalAccessTokenManager(self)
"""See :class:`~gitlab.v4.objects.PersonalAccessTokenManager`"""
self.topics = objects.TopicManager(self)
"""See :class:`~gitlab.v4.objects.TopicManager`"""
def __enter__(self) -> "Gitlab":
return self
def __exit__(self, *args: Any) -> None:
self.session.close()
def __getstate__(self) -> Dict[str, Any]:
state = self.__dict__.copy()
state.pop("_objects")
return state
def __setstate__(self, state: Dict[str, Any]) -> None:
self.__dict__.update(state)
# We only support v4 API at this time
if self._api_version not in ("4",):
raise ModuleNotFoundError(name=f"gitlab.v{self._api_version}.objects")
# NOTE: We must delay import of gitlab.v4.objects until now or
# otherwise it will cause circular import errors
import gitlab.v4.objects
self._objects = gitlab.v4.objects
@property
def url(self) -> str:
"""The user-provided server URL."""
return self._base_url
@property
def api_url(self) -> str:
"""The computed API base URL."""
return self._url
@property
def api_version(self) -> str:
"""The API version used (4 only)."""
return self._api_version
@classmethod
def from_config(
cls, gitlab_id: Optional[str] = None, config_files: Optional[List[str]] = None
) -> "Gitlab":
"""Create a Gitlab connection from configuration files.
Args:
gitlab_id: ID of the configuration section.
config_files list[str]: List of paths to configuration files.
Returns:
A Gitlab connection.
Raises:
gitlab.config.GitlabDataError: If the configuration is not correct.
"""
config = gitlab.config.GitlabConfigParser(
gitlab_id=gitlab_id, config_files=config_files
)
return cls(
config.url,
private_token=config.private_token,
oauth_token=config.oauth_token,
job_token=config.job_token,
ssl_verify=config.ssl_verify,
timeout=config.timeout,
http_username=config.http_username,
http_password=config.http_password,
api_version=config.api_version,
per_page=config.per_page,
pagination=config.pagination,
order_by=config.order_by,
user_agent=config.user_agent,
retry_transient_errors=config.retry_transient_errors,
)
@classmethod
def merge_config(
cls,
options: dict,
gitlab_id: Optional[str] = None,
config_files: Optional[List[str]] = None,
) -> "Gitlab":
"""Create a Gitlab connection by merging configuration with
the following precedence:
1. Explicitly provided CLI arguments,
2. Environment variables,
3. Configuration files:
a. explicitly defined config files:
i. via the `--config-file` CLI argument,
ii. via the `PYTHON_GITLAB_CFG` environment variable,
b. user-specific config file,
c. system-level config file,
4. Environment variables always present in CI (CI_SERVER_URL, CI_JOB_TOKEN).
Args:
options: A dictionary of explicitly provided key-value options.
gitlab_id: ID of the configuration section.
config_files: List of paths to configuration files.
Returns:
(gitlab.Gitlab): A Gitlab connection.
Raises:
gitlab.config.GitlabDataError: If the configuration is not correct.
"""
config = gitlab.config.GitlabConfigParser(
gitlab_id=gitlab_id, config_files=config_files
)
url = (
options.get("server_url")
or config.url
or os.getenv("CI_SERVER_URL")
or gitlab.const.DEFAULT_URL
)
private_token, oauth_token, job_token = cls._merge_auth(options, config)
return cls(
url=url,
private_token=private_token,
oauth_token=oauth_token,
job_token=job_token,
ssl_verify=options.get("ssl_verify") or config.ssl_verify,
timeout=options.get("timeout") or config.timeout,
api_version=options.get("api_version") or config.api_version,
per_page=options.get("per_page") or config.per_page,
pagination=options.get("pagination") or config.pagination,
order_by=options.get("order_by") or config.order_by,
user_agent=options.get("user_agent") or config.user_agent,
)
@staticmethod
def _merge_auth(options: dict, config: gitlab.config.GitlabConfigParser) -> Tuple:
"""
Return a tuple where at most one of 3 token types ever has a value.
Since multiple types of tokens may be present in the environment,
options, or config files, this precedence ensures we don't
inadvertently cause errors when initializing the client.
This is especially relevant when executed in CI where user and
CI-provided values are both available.
"""
private_token = options.get("private_token") or config.private_token
oauth_token = options.get("oauth_token") or config.oauth_token
job_token = (
options.get("job_token") or config.job_token or os.getenv("CI_JOB_TOKEN")
)
if private_token:
return (private_token, None, None)
if oauth_token:
return (None, oauth_token, None)
if job_token:
return (None, None, job_token)
return (None, None, None)
def auth(self) -> None:
"""Performs an authentication using private token.
The `user` attribute will hold a `gitlab.objects.CurrentUser` object on
success.
"""
self.user = self._objects.CurrentUserManager(self).get()
def version(self) -> Tuple[str, str]:
"""Returns the version and revision of the gitlab server.
Note that self.version and self.revision will be set on the gitlab
object.
Returns:
The server version and server revision.
('unknown', 'unknown') if the server doesn't perform as expected.
"""
if self._server_version is None:
try:
data = self.http_get("/version")
if isinstance(data, dict):
self._server_version = data["version"]
self._server_revision = data["revision"]
else:
self._server_version = "unknown"
self._server_revision = "unknown"
except Exception:
self._server_version = "unknown"
self._server_revision = "unknown"
return cast(str, self._server_version), cast(str, self._server_revision)
@gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabVerifyError)
def lint(self, content: str, **kwargs: Any) -> Tuple[bool, List[str]]:
"""Validate a gitlab CI configuration.
Args:
content: The .gitlab-ci.yml content
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabVerifyError: If the validation could not be done
Returns:
(True, []) if the file is valid, (False, errors(list)) otherwise
"""
post_data = {"content": content}
data = self.http_post("/ci/lint", post_data=post_data, **kwargs)
if TYPE_CHECKING:
assert not isinstance(data, requests.Response)
return (data["status"] == "valid", data["errors"])
@gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabMarkdownError)
def markdown(
self, text: str, gfm: bool = False, project: Optional[str] = None, **kwargs: Any
) -> str:
"""Render an arbitrary Markdown document.
Args:
text: The markdown text to render
gfm: Render text using GitLab Flavored Markdown. Default is False
project: Full path of a project used a context when `gfm` is True
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabMarkdownError: If the server cannot perform the request
Returns:
The HTML rendering of the markdown text.
"""
post_data = {"text": text, "gfm": gfm}
if project is not None:
post_data["project"] = project
data = self.http_post("/markdown", post_data=post_data, **kwargs)
if TYPE_CHECKING:
assert not isinstance(data, requests.Response)
return data["html"]
@gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabLicenseError)
def get_license(self, **kwargs: Any) -> Dict[str, Any]:
"""Retrieve information about the current license.
Args:
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabGetError: If the server cannot perform the request
Returns:
The current license information
"""
result = self.http_get("/license", **kwargs)
if isinstance(result, dict):
return result
return {}
@gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabLicenseError)
def set_license(self, license: str, **kwargs: Any) -> Dict[str, Any]:
"""Add a new license.
Args:
license: The license string
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabPostError: If the server cannot perform the request
Returns:
The new license information
"""
data = {"license": license}
result = self.http_post("/license", post_data=data, **kwargs)
if TYPE_CHECKING:
assert not isinstance(result, requests.Response)
return result
def _set_auth_info(self) -> None:
tokens = [
token
for token in [self.private_token, self.oauth_token, self.job_token]
if token
]
if len(tokens) > 1:
raise ValueError(
"Only one of private_token, oauth_token or job_token should "
"be defined"
)
if (self.http_username and not self.http_password) or (
not self.http_username and self.http_password
):
raise ValueError(
"Both http_username and http_password should " "be defined"
)
if self.oauth_token and self.http_username:
raise ValueError(
"Only one of oauth authentication or http "
"authentication should be defined"
)
self._http_auth = None
if self.private_token:
self.headers.pop("Authorization", None)
self.headers["PRIVATE-TOKEN"] = self.private_token
self.headers.pop("JOB-TOKEN", None)
if self.oauth_token:
self.headers["Authorization"] = f"Bearer {self.oauth_token}"
self.headers.pop("PRIVATE-TOKEN", None)
self.headers.pop("JOB-TOKEN", None)
if self.job_token:
self.headers.pop("Authorization", None)
self.headers.pop("PRIVATE-TOKEN", None)
self.headers["JOB-TOKEN"] = self.job_token
if self.http_username:
self._http_auth = requests.auth.HTTPBasicAuth(
self.http_username, self.http_password
)
def enable_debug(self) -> None:
import logging
from http.client import HTTPConnection # noqa
HTTPConnection.debuglevel = 1
logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
def _get_session_opts(self) -> Dict[str, Any]:
return {
"headers": self.headers.copy(),
"auth": self._http_auth,
"timeout": self.timeout,
"verify": self.ssl_verify,
}
def _get_base_url(self, url: Optional[str] = None) -> str:
"""Return the base URL with the trailing slash stripped.
If the URL is a Falsy value, return the default URL.
Returns:
The base URL
"""
if not url:
return gitlab.const.DEFAULT_URL
return url.rstrip("/")
def _build_url(self, path: str) -> str:
"""Returns the full url from path.
If path is already a url, return it unchanged. If it's a path, append
it to the stored url.
Returns:
The full URL
"""
if path.startswith("http://") or path.startswith("https://"):
return path
else:
return f"{self._url}{path}"
def _check_redirects(self, result: requests.Response) -> None:
# Check the requests history to detect 301/302 redirections.
# If the initial verb is POST or PUT, the redirected request will use a
# GET request, leading to unwanted behaviour.
# If we detect a redirection with a POST or a PUT request, we
# raise an exception with a useful error message.
if not result.history:
return
for item in result.history:
if item.status_code not in (301, 302):
continue
# GET methods can be redirected without issue
if item.request.method == "GET":
continue
target = item.headers.get("location")
raise gitlab.exceptions.RedirectError(
REDIRECT_MSG.format(
status_code=item.status_code,
reason=item.reason,
source=item.url,
target=target,
)
)
def _prepare_send_data(
self,
files: Optional[Dict[str, Any]] = None,
post_data: Optional[Union[Dict[str, Any], bytes]] = None,
raw: bool = False,
) -> Tuple[
Optional[Union[Dict[str, Any], bytes]],
Optional[Union[Dict[str, Any], MultipartEncoder]],
str,
]:
if files:
if post_data is None:
post_data = {}
else:
# booleans does not exists for data (neither for MultipartEncoder):
# cast to string int to avoid: 'bool' object has no attribute 'encode'
if TYPE_CHECKING:
assert isinstance(post_data, dict)
for k, v in post_data.items():
if isinstance(v, bool):
post_data[k] = str(int(v))
post_data["file"] = files.get("file")
post_data["avatar"] = files.get("avatar")
data = MultipartEncoder(post_data)
return (None, data, data.content_type)
if raw and post_data:
return (None, post_data, "application/octet-stream")
return (post_data, None, "application/json")
def http_request(
self,
verb: str,
path: str,
query_data: Optional[Dict[str, Any]] = None,
post_data: Optional[Union[Dict[str, Any], bytes]] = None,
raw: bool = False,
streamed: bool = False,
files: Optional[Dict[str, Any]] = None,
timeout: Optional[float] = None,
obey_rate_limit: bool = True,
max_retries: int = 10,
**kwargs: Any,
) -> requests.Response:
"""Make an HTTP request to the Gitlab server.
Args:
verb: The HTTP method to call ('get', 'post', 'put', 'delete')
path: Path or full URL to query ('/projects' or
'http://whatever/v4/api/projecs')
query_data: Data to send as query parameters
post_data: Data to send in the body (will be converted to
json by default)
raw: If True, do not convert post_data to json
streamed: Whether the data should be streamed
files: The files to send to the server
timeout: The timeout, in seconds, for the request
obey_rate_limit: Whether to obey 429 Too Many Request
responses. Defaults to True.
max_retries: Max retries after 429 or transient errors,
set to -1 to retry forever. Defaults to 10.
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
A requests result object.
Raises:
GitlabHttpError: When the return code is not 2xx
"""
query_data = query_data or {}
url = self._build_url(path)
params: Dict[str, Any] = {}
utils.copy_dict(params, query_data)
# Deal with kwargs: by default a user uses kwargs to send data to the
# gitlab server, but this generates problems (python keyword conflicts
# and python-gitlab/gitlab conflicts).
# So we provide a `query_parameters` key: if it's there we use its dict
# value as arguments for the gitlab server, and ignore the other
# arguments, except pagination ones (per_page and page)
if "query_parameters" in kwargs:
utils.copy_dict(params, kwargs["query_parameters"])
for arg in ("per_page", "page"):
if arg in kwargs:
params[arg] = kwargs[arg]
else:
utils.copy_dict(params, kwargs)
opts = self._get_session_opts()
verify = opts.pop("verify")
opts_timeout = opts.pop("timeout")
# If timeout was passed into kwargs, allow it to override the default
if timeout is None:
timeout = opts_timeout
# We need to deal with json vs. data when uploading files
json, data, content_type = self._prepare_send_data(files, post_data, raw)
opts["headers"]["Content-type"] = content_type
cur_retries = 0
while True:
result = self.session.request(
method=verb,
url=url,
json=json,
data=data,
params=params,
timeout=timeout,
verify=verify,
stream=streamed,
**opts,
)
self._check_redirects(result)
if 200 Union[Dict[str, Any], requests.Response]:
"""Make a GET request to the Gitlab server.
Args:
path: Path or full URL to query ('/projects' or
'http://whatever/v4/api/projecs')
query_data: Data to send as query parameters
streamed: Whether the data should be streamed
raw: If True do not try to parse the output as json
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
A requests result object is streamed is True or the content type is
not json.
The parsed json data otherwise.
Raises:
GitlabHttpError: When the return code is not 2xx
GitlabParsingError: If the json data could not be parsed
"""
query_data = query_data or {}
result = self.http_request(
"get", path, query_data=query_data, streamed=streamed, **kwargs
)
if (
result.headers["Content-Type"] == "application/json"
and not streamed
and not raw
):
try:
return result.json()
except Exception as e:
raise gitlab.exceptions.GitlabParsingError(
error_message="Failed to parse the server message"
) from e
else:
return result
def http_list(
self,
path: str,
query_data: Optional[Dict[str, Any]] = None,
as_list: Optional[bool] = None,
**kwargs: Any,
) -> Union["GitlabList", List[Dict[str, Any]]]:
"""Make a GET request to the Gitlab server for list-oriented queries.
Args:
path: Path or full URL to query ('/projects' or
'http://whatever/v4/api/projects')
query_data: Data to send as query parameters
**kwargs: Extra options to send to the server (e.g. sudo, page,
per_page)
Returns:
A list of the objects returned by the server. If `as_list` is
False and no pagination-related arguments (`page`, `per_page`,
`all`) are defined then a GitlabList object (generator) is returned
instead. This object will make API calls when needed to fetch the
next items from the server.
Raises:
GitlabHttpError: When the return code is not 2xx
GitlabParsingError: If the json data could not be parsed
"""
query_data = query_data or {}
# In case we want to change the default behavior at some point
as_list = True if as_list is None else as_list
get_all = kwargs.pop("all", False)
url = self._build_url(path)
page = kwargs.get("page")
if get_all is True and as_list is True:
return list(GitlabList(self, url, query_data, **kwargs))
if page or as_list is True:
# pagination requested, we return a list
return list(GitlabList(self, url, query_data, get_next=False, **kwargs))
# No pagination, generator requested
return GitlabList(self, url, query_data, **kwargs)
def http_post(
self,
path: str,
query_data: Optional[Dict[str, Any]] = None,
post_data: Optional[Dict[str, Any]] = None,
raw: bool = False,
files: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> Union[Dict[str, Any], requests.Response]:
"""Make a POST request to the Gitlab server.
Args:
path: Path or full URL to query ('/projects' or
'http://whatever/v4/api/projecs')
query_data: Data to send as query parameters
post_data: Data to send in the body (will be converted to
json by default)
raw: If True, do not convert post_data to json
files: The files to send to the server
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
The parsed json returned by the server if json is return, else the
raw content
Raises:
GitlabHttpError: When the return code is not 2xx
GitlabParsingError: If the json data could not be parsed
"""
query_data = query_data or {}
post_data = post_data or {}
result = self.http_request(
"post",
path,
query_data=query_data,
post_data=post_data,
files=files,
**kwargs,
)
try:
if result.headers.get("Content-Type", None) == "application/json":
return result.json()
except Exception as e:
raise gitlab.exceptions.GitlabParsingError(
error_message="Failed to parse the server message"
) from e
return result
def http_put(
self,
path: str,
query_data: Optional[Dict[str, Any]] = None,
post_data: Optional[Union[Dict[str, Any], bytes]] = None,
raw: bool = False,
files: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> Union[Dict[str, Any], requests.Response]:
"""Make a PUT request to the Gitlab server.
Args:
path: Path or full URL to query ('/projects' or
'http://whatever/v4/api/projecs')
query_data: Data to send as query parameters
post_data: Data to send in the body (will be converted to
json by default)
raw: If True, do not convert post_data to json
files: The files to send to the server
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
The parsed json returned by the server.
Raises:
GitlabHttpError: When the return code is not 2xx
GitlabParsingError: If the json data could not be parsed
"""
query_data = query_data or {}
post_data = post_data or {}
result = self.http_request(
"put",
path,
query_data=query_data,
post_data=post_data,
files=files,
raw=raw,
**kwargs,
)
try:
return result.json()
except Exception as e:
raise gitlab.exceptions.GitlabParsingError(
error_message="Failed to parse the server message"
) from e
def http_delete(self, path: str, **kwargs: Any) -> requests.Response:
"""Make a DELETE request to the Gitlab server.
Args:
path: Path or full URL to query ('/projects' or
'http://whatever/v4/api/projecs')
**kwargs: Extra options to send to the server (e.g. sudo)
Returns:
The requests object.
Raises:
GitlabHttpError: When the return code is not 2xx
"""
return self.http_request("delete", path, **kwargs)
@gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabSearchError)
def search(
self, scope: str, search: str, **kwargs: Any
) -> Union["GitlabList", List[Dict[str, Any]]]:
"""Search GitLab resources matching the provided string.'
Args:
scope: Scope of the search
search: Search string
**kwargs: Extra options to send to the server (e.g. sudo)
Raises:
GitlabAuthenticationError: If authentication is not correct
GitlabSearchError: If the server failed to perform the request
Returns:
A list of dicts describing the resources found.
"""
data = {"scope": scope, "search": search}
return self.http_list("/search", query_data=data, **kwargs)
class GitlabList(object):
"""Generator representing a list of remote objects.
The object handles the links returned by a query to the API, and will call
the API again when needed.
"""
def __init__(
self,
gl: Gitlab,
url: str,
query_data: Dict[str, Any],
get_next: bool = True,
**kwargs: Any,
) -> None:
self._gl = gl
# Preserve kwargs for subsequent queries
self._kwargs = kwargs.copy()
self._query(url, query_data, **self._kwargs)
self._get_next = get_next
# Remove query_parameters from kwargs, which are saved via the `next` URL
self._kwargs.pop("query_parameters", None)
def _query(
self, url: str, query_data: Optional[Dict[str, Any]] = None, **kwargs: Any
) -> None:
query_data = query_data or {}
result = self._gl.http_request("get", url, query_data=query_data, **kwargs)
try:
links = result.links
if links:
next_url = links["next"]["url"]
else:
next_url = requests.utils.parse_header_links(result.headers["links"])[
0
]["url"]
self._next_url = next_url
except KeyError:
self._next_url = None
self._current_page: Optional[str] = result.headers.get("X-Page")
self._prev_page: Optional[str] = result.headers.get("X-Prev-Page")
self._next_page: Optional[str] = result.headers.get("X-Next-Page")
self._per_page: Optional[str] = result.headers.get("X-Per-Page")
self._total_pages: Optional[str] = result.headers.get("X-Total-Pages")
self._total: Optional[str] = result.headers.get("X-Total")
try:
self._data: List[Dict[str, Any]] = result.json()
except Exception as e:
raise gitlab.exceptions.GitlabParsingError(
error_message="Failed to parse the server message"
) from e
self._current = 0
@property
def current_page(self) -> int:
"""The current page number."""
if TYPE_CHECKING:
assert self._current_page is not None
return int(self._current_page)
@property
def prev_page(self) -> Optional[int]:
"""The previous page number.
If None, the current page is the first.
"""
return int(self._prev_page) if self._prev_page else None
@property
def next_page(self) -> Optional[int]:
"""The next page number.
If None, the current page is the last.
"""
return int(self._next_page) if self._next_page else None
@property
def per_page(self) -> int:
"""The number of items per page."""
if TYPE_CHECKING:
assert self._per_page is not None
return int(self._per_page)
# NOTE(jlvillal): When a query returns more than 10,000 items, GitLab doesn't return
# the headers 'x-total-pages' and 'x-total'. In those cases we return None.
# https://docs.gitlab.com/ee/user/gitlab_com/index.html#pagination-response-headers
@property
def total_pages(self) -> Optional[int]:
"""The total number of pages."""
if self._total_pages is not None:
return int(self._total_pages)
return None
@property
def total(self) -> Optional[int]:
"""The total number of items."""
if self._total is not None:
return int(self._total)
return None
def __iter__(self) -> "GitlabList":
return self
def __len__(self) -> int:
if self._total is None:
return 0
return int(self._total)
def __next__(self) -> Dict[str, Any]:
return self.next()
def next(self) -> Dict[str, Any]:
try:
item = self._data[self._current]
self._current += 1
return item
except IndexError:
pass
if self._next_url and self._get_next is True:
self._query(self._next_url, **self._kwargs)
return self.next()
raise StopIteration