"""Wrapper for the GitLab API.""" import os import re import time from typing import Any, cast, Dict, List, Optional, Tuple, TYPE_CHECKING, Union from urllib import parse import requests import requests.utils from requests_toolbelt.multipart.encoder import MultipartEncoder # type: ignore import gitlab 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}" ) RETRYABLE_TRANSIENT_ERROR_CODES = [500, 502, 503, 504] + list(range(520, 531)) # https://docs.gitlab.com/ee/api/#offset-based-pagination _PAGINATION_URL = ( f"https://python-gitlab.readthedocs.io/en/v{gitlab.__version__}/" f"api-usage.html#pagination" ) class Gitlab: """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, 504 or 52x responses. Defaults to False. keep_base_url: keep user-provided base URL for pagination if it differs from response headers """ 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, keep_base_url: 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 self.keep_base_url = keep_base_url #: 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(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 from gitlab.v4 import objects self._objects = objects self.user: Optional[objects.CurrentUser] = None self.broadcastmessages = objects.BroadcastMessageManager(self) """See :class:`~gitlab.v4.objects.BroadcastMessageManager`""" self.ci_lint = objects.CiLintManager(self) """See :class:`~gitlab.v4.objects.CiLintManager`""" 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.registry_repositories = objects.RegistryRepositoryManager(self) """See :class:`~gitlab.v4.objects.RegistryRepositoryManager`""" self.runners = objects.RunnerManager(self) """See :class:`~gitlab.v4.objects.RunnerManager`""" self.runners_all = objects.RunnerAllManager(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`""" self.statistics = objects.ApplicationStatisticsManager(self) """See :class:`~gitlab.v4.objects.ApplicationStatisticsManager`""" 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( f"gitlab.v{self._api_version}.objects" ) # pragma: no cover, dead code currently # NOTE: We must delay import of gitlab.v4.objects until now or # otherwise it will cause circular import errors from gitlab.v4 import objects self._objects = 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[str, Any], 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[str, Any], config: gitlab.config.GitlabConfigParser ) -> Tuple[Optional[str], Optional[str], Optional[str]]: """ 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. Warns the user if a potentially misconfigured URL is detected on the client or server side. The `user` attribute will hold a `gitlab.objects.CurrentUser` object on success. """ self.user = self._objects.CurrentUserManager(self).get() if hasattr(self.user, "web_url") and hasattr(self.user, "username"): self._check_url(self.user.web_url, path=self.user.username) 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 """ utils.warn( "`lint()` is deprecated and will be removed in a future version.\n" "Please use `ci_lint.create()` instead.", category=DeprecationWarning, ) 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) assert isinstance(data["html"], str) return data["html"] @gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabLicenseError) def get_license(self, **kwargs: Any) -> Dict[str, Union[str, Dict[str, str]]]: """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 ) @staticmethod def enable_debug() -> 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, } @staticmethod def _get_base_url(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 return f"{self._url}{path}" def _check_url(self, url: Optional[str], *, path: str = "api") -> Optional[str]: """ Checks if ``url`` starts with a different base URL from the user-provided base URL and warns the user before returning it. If ``keep_base_url`` is set to ``True``, instead returns the URL massaged to match the user-provided base URL. """ if not url or url.startswith(self.url): return url match = re.match(rf"(^.*?)/{path}", url) if not match: return url base_url = match.group(1) if self.keep_base_url: return url.replace(base_url, f"{self._base_url}") utils.warn( message=( f"The base URL in the server response differs from the user-provided " f"base URL ({self.url} -> {base_url}).\nThis is usually caused by a " f"misconfigured base URL on your side or a misconfigured external_url " f"on the server side, and can lead to broken pagination and unexpected " f"behavior. If this is intentional, use `keep_base_url=True` when " f"initializing the Gitlab instance to keep the user-provided base URL." ), category=UserWarning, ) return url @staticmethod def _check_redirects(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, ) ) @staticmethod def _prepare_send_data( 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, retry_transient_errors: Optional[bool] = None, 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. retry_transient_errors: Whether to retry after 500, 502, 503, 504 or 52x responses. Defaults to False. 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 {} raw_url = self._build_url(path) # parse user-provided URL params to ensure we don't add our own duplicates parsed = parse.urlparse(raw_url) params = parse.parse_qs(parsed.query) utils.copy_dict(src=query_data, dest=params) url = parse.urlunparse(parsed._replace(query="")) # 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(src=kwargs["query_parameters"], dest=params) for arg in ("per_page", "page"): if arg in kwargs: params[arg] = kwargs[arg] else: utils.copy_dict(src=kwargs, dest=params) 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 if retry_transient_errors is None: retry_transient_errors = self.retry_transient_errors # 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: try: result = self.session.request( method=verb, url=url, json=json, data=data, params=params, timeout=timeout, verify=verify, stream=streamed, **opts, ) except (requests.ConnectionError, requests.exceptions.ChunkedEncodingError): if retry_transient_errors and ( max_retries == -1 or cur_retries < max_retries ): wait_time = 2**cur_retries * 0.1 cur_retries += 1 time.sleep(wait_time) continue raise self._check_redirects(result) if 200 <= result.status_code < 300: return result if (429 == result.status_code and obey_rate_limit) or ( result.status_code in RETRYABLE_TRANSIENT_ERROR_CODES and retry_transient_errors ): # Response headers documentation: # https://docs.gitlab.com/ee/user/admin_area/settings/user_and_ip_rate_limits.html#response-headers 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"]) elif "RateLimit-Reset" in result.headers: wait_time = int(result.headers["RateLimit-Reset"]) - time.time() 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: 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: json_result = result.json() if TYPE_CHECKING: assert isinstance(json_result, dict) return json_result except Exception as e: raise gitlab.exceptions.GitlabParsingError( error_message="Failed to parse the server message" ) from e else: return result def http_head( self, path: str, query_data: Optional[Dict[str, Any]] = None, **kwargs: Any ) -> "requests.structures.CaseInsensitiveDict[Any]": """Make a HEAD 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 **kwargs: Extra options to send to the server (e.g. sudo, page, per_page) Returns: A requests.header object Raises: GitlabHttpError: When the return code is not 2xx """ query_data = query_data or {} result = self.http_request("head", path, query_data=query_data, **kwargs) return result.headers def http_list( self, path: str, query_data: Optional[Dict[str, Any]] = None, *, as_list: Optional[bool] = None, # Deprecated in favor of `iterator` iterator: 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 iterator: Indicate if should return a generator (True) **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 `iterator` is True and no pagination-related arguments (`page`, `per_page`, `get_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 {} # Don't allow both `as_list` and `iterator` to be set. if as_list is not None and iterator is not None: raise ValueError( "Only one of `as_list` or `iterator` can be used. " "Use `iterator` instead of `as_list`. `as_list` is deprecated." ) if as_list is not None: iterator = not as_list utils.warn( message=( f"`as_list={as_list}` is deprecated and will be removed in a " f"future version. Use `iterator={iterator}` instead." ), category=DeprecationWarning, ) # Provide a `get_all`` param to avoid clashes with `all` API attributes. get_all = kwargs.pop("get_all", None) if get_all is None: # For now, keep `all` without deprecation. get_all = kwargs.pop("all", None) url = self._build_url(path) page = kwargs.get("page") if iterator: # Generator requested return GitlabList(self, url, query_data, **kwargs) if get_all is True: return list(GitlabList(self, url, query_data, **kwargs)) # pagination requested, we return a list gl_list = GitlabList(self, url, query_data, get_next=False, **kwargs) items = list(gl_list) def should_emit_warning() -> bool: # No warning is emitted if any of the following conditions apply: # * `get_all=False` was set in the `list()` call. # * `page` was set in the `list()` call. # * GitLab did not return the `x-per-page` header. # * Number of items received is less than per-page value. # * Number of items received is >= total available. if get_all is False: return False if page is not None: return False if gl_list.per_page is None: return False if len(items) < gl_list.per_page: return False if gl_list.total is not None and len(items) >= gl_list.total: return False return True if not should_emit_warning(): return items # Warn the user that they are only going to retrieve `per_page` # maximum items. This is a common cause of issues filed. total_items = "many" if gl_list.total is None else gl_list.total utils.warn( message=( f"Calling a `list()` method without specifying `get_all=True` or " f"`iterator=True` will return a maximum of {gl_list.per_page} items. " f"Your query returned {len(items)} of {total_items} items. See " f"{_PAGINATION_URL} for more details. If this was done intentionally, " f"then this warning can be supressed by adding the argument " f"`get_all=False` to the `list()` call." ), category=UserWarning, ) return items 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, raw=raw, **kwargs, ) try: if result.headers.get("Content-Type", None) == "application/json": json_result = result.json() if TYPE_CHECKING: assert isinstance(json_result, dict) return json_result 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: json_result = result.json() if TYPE_CHECKING: assert isinstance(json_result, dict) return json_result 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: """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: next_url = result.links["next"]["url"] except KeyError: next_url = None self._next_url = self._gl._check_url(next_url) 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) -> Optional[int]: """The number of items per page.""" return int(self._per_page) if self._per_page is not None else None # 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