diff options
author | Nejc Habjan <hab.nejc@gmail.com> | 2021-12-01 01:04:53 +0100 |
---|---|---|
committer | GitHub <noreply@github.com> | 2021-12-01 01:04:53 +0100 |
commit | 8d76826fa64460e504acc5924f859f8dbc246b42 (patch) | |
tree | 083fefada982c795e2415092794db429abb0c184 /gitlab/client.py | |
parent | 5a1678f43184bd459132102cc13cf8426fe0449d (diff) | |
parent | 86ab04e54ea4175f10053decfad5086cda7aa024 (diff) | |
download | gitlab-master.tar.gz |
Merge pull request #1723 from python-gitlab/jlvillal/dead_mastermaster
Close-out `master` branch
Diffstat (limited to 'gitlab/client.py')
-rw-r--r-- | gitlab/client.py | 1011 |
1 files changed, 0 insertions, 1011 deletions
diff --git a/gitlab/client.py b/gitlab/client.py deleted file mode 100644 index 8bec64f..0000000 --- a/gitlab/client.py +++ /dev/null @@ -1,1011 +0,0 @@ -# -*- coding: utf-8 -*- -# -# Copyright (C) 2013-2017 Gauvain Pocentek <gauvain@pocentek.net> -# -# 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 <http://www.gnu.org/licenses/>. -"""Wrapper for the GitLab API.""" - -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 (str): The URL of the GitLab server (defaults to https://gitlab.com). - private_token (str): The user private token - oauth_token (str): An oauth token - job_token (str): A CI job token - ssl_verify (bool|str): 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 (float): Timeout to use for requests to the GitLab server. - http_username (str): Username for HTTP authentication - http_password (str): Password for HTTP authentication - api_version (str): Gitlab API version to use (support for 4 only) - pagination (str): Can be set to 'keyset' to use keyset pagination - order_by (str): Set order_by globally - user_agent (str): A custom user agent to use for making HTTP requests. - retry_transient_errors (bool): 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 = "%s/api/v%s" % (self._base_url, 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="gitlab.v%s.objects" % self._api_version) - # 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`""" - - 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="gitlab.v%s.objects" % self._api_version) - # 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 (str): ID of the configuration section. - config_files list[str]: 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 - ) - 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, - ) - - 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: - tuple (str, str): The server version and server revision. - ('unknown', 'unknwown') 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 (txt): 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: - tuple: (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 (str): The markdown text to render - gfm (bool): Render text using GitLab Flavored Markdown. Default is - False - project (str): 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: - str: 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: - dict: 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 (str): 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: - dict: 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"] = "Bearer %s" % 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 # type: ignore - 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: - str: 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: - str: The full URL - """ - if path.startswith("http://") or path.startswith("https://"): - return path - else: - return "%s%s" % (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[Dict[str, Any]] = None, - raw: bool = False, - ) -> Tuple[ - Optional[Dict[str, Any]], - 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' - 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[Dict[str, Any]] = 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 (str): The HTTP method to call ('get', 'post', 'put', - 'delete') - path (str): Path or full URL to query ('/projects' or - 'http://whatever/v4/api/projecs') - query_data (dict): Data to send as query parameters - post_data (dict): Data to send in the body (will be converted to - json by default) - raw (bool): If True, do not convert post_data to json - streamed (bool): Whether the data should be streamed - files (dict): The files to send to the server - timeout (float): The timeout, in seconds, for the request - obey_rate_limit (bool): Whether to obey 429 Too Many Request - responses. Defaults to True. - max_retries (int): 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 - - # Requests assumes that `.` should not be encoded as %2E and will make - # changes to urls using this encoding. Using a prepped request we can - # get the desired behavior. - # The Requests behavior is right but it seems that web servers don't - # always agree with this decision (this is the case with a default - # gitlab installation) - req = requests.Request(verb, url, json=json, data=data, params=params, **opts) - prepped = self.session.prepare_request(req) - if TYPE_CHECKING: - assert prepped.url is not None - prepped.url = utils.sanitized_url(prepped.url) - settings = self.session.merge_environment_settings( - prepped.url, {}, streamed, verify, None - ) - - cur_retries = 0 - while True: - result = self.session.send(prepped, timeout=timeout, **settings) - - self._check_redirects(result) - - if 200 <= result.status_code < 300: - return result - - retry_transient_errors = kwargs.get( - "retry_transient_errors", self.retry_transient_errors - ) - if (429 == result.status_code and obey_rate_limit) or ( - result.status_code in [500, 502, 503, 504] and retry_transient_errors - ): - if max_retries == -1 or cur_retries < max_retries: - wait_time = 2 ** cur_retries * 0.1 - if "Retry-After" in result.headers: - wait_time = int(result.headers["Retry-After"]) - cur_retries += 1 - time.sleep(wait_time) - continue - - error_message = result.content - try: - error_json = result.json() - for k in ("message", "error"): - if k in error_json: - error_message = error_json[k] - except (KeyError, ValueError, TypeError): - pass - - if result.status_code == 401: - raise gitlab.exceptions.GitlabAuthenticationError( - response_code=result.status_code, - error_message=error_message, - response_body=result.content, - ) - - raise gitlab.exceptions.GitlabHttpError( - response_code=result.status_code, - error_message=error_message, - response_body=result.content, - ) - - def http_get( - self, - path: str, - query_data: Optional[Dict[str, Any]] = None, - streamed: bool = False, - raw: bool = False, - **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: - """Make a GET request to the Gitlab server. - - Args: - path (str): Path or full URL to query ('/projects' or - 'http://whatever/v4/api/projecs') - query_data (dict): Data to send as query parameters - streamed (bool): Whether the data should be streamed - raw (bool): 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 (str): Path or full URL to query ('/projects' or - 'http://whatever/v4/api/projects') - query_data (dict): Data to send as query parameters - **kwargs: Extra options to send to the server (e.g. sudo, page, - per_page) - - Returns: - list: 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 (str): Path or full URL to query ('/projects' or - 'http://whatever/v4/api/projecs') - query_data (dict): Data to send as query parameters - post_data (dict): Data to send in the body (will be converted to - json by default) - raw (bool): If True, do not convert post_data to json - files (dict): 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[Dict[str, Any]] = 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 (str): Path or full URL to query ('/projects' or - 'http://whatever/v4/api/projecs') - query_data (dict): Data to send as query parameters - post_data (dict): Data to send in the body (will be converted to - json by default) - raw (bool): If True, do not convert post_data to json - files (dict): 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 (str): 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 (str): Scope of the search - search (str): 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: - GitlabList: 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[Union[str, int]] = result.headers.get("X-Page") - self._prev_page: Optional[Union[str, int]] = result.headers.get("X-Prev-Page") - self._next_page: Optional[Union[str, int]] = result.headers.get("X-Next-Page") - self._per_page: Optional[Union[str, int]] = result.headers.get("X-Per-Page") - self._total_pages: Optional[Union[str, int]] = result.headers.get( - "X-Total-Pages" - ) - self._total: Optional[Union[str, int]] = 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) - - @property - def total_pages(self) -> int: - """The total number of pages.""" - if TYPE_CHECKING: - assert self._total_pages is not None - return int(self._total_pages) - - @property - def total(self) -> int: - """The total number of items.""" - if TYPE_CHECKING: - assert self._total is not None - return int(self._total) - - 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 |