import configparser
import os
import shlex
import subprocess
from os.path import expanduser, expandvars
from pathlib import Path
from typing import List, Optional, Union
from gitlab.const import USER_AGENT
_DEFAULT_FILES: List[str] = [
"/etc/python-gitlab.cfg",
str(Path.home() / ".python-gitlab.cfg"),
]
HELPER_PREFIX = "helper:"
HELPER_ATTRIBUTES = ["job_token", "http_password", "private_token", "oauth_token"]
_CONFIG_PARSER_ERRORS = (configparser.NoOptionError, configparser.NoSectionError)
def _resolve_file(filepath: Union[Path, str]) -> str:
resolved = Path(filepath).resolve(strict=True)
return str(resolved)
def _get_config_files(
config_files: Optional[List[str]] = None,
) -> Union[str, List[str]]:
"""
Return resolved path(s) to config files if they exist, with precedence:
1. Files passed in config_files
2. File defined in PYTHON_GITLAB_CFG
3. User- and system-wide config files
"""
resolved_files = []
if config_files:
for config_file in config_files:
try:
resolved = _resolve_file(config_file)
except OSError as e:
raise GitlabConfigMissingError(
f"Cannot read config from file: {e}"
) from e
resolved_files.append(resolved)
return resolved_files
try:
env_config = os.environ["PYTHON_GITLAB_CFG"]
return _resolve_file(env_config)
except KeyError:
pass
except OSError as e:
raise GitlabConfigMissingError(
f"Cannot read config from PYTHON_GITLAB_CFG: {e}"
) from e
for config_file in _DEFAULT_FILES:
try:
resolved = _resolve_file(config_file)
except OSError:
continue
resolved_files.append(resolved)
return resolved_files
class ConfigError(Exception):
pass
class GitlabIDError(ConfigError):
pass
class GitlabDataError(ConfigError):
pass
class GitlabConfigMissingError(ConfigError):
pass
class GitlabConfigHelperError(ConfigError):
pass
class GitlabConfigParser:
def __init__(
self, gitlab_id: Optional[str] = None, config_files: Optional[List[str]] = None
) -> None:
self.gitlab_id = gitlab_id
self.http_username: Optional[str] = None
self.http_password: Optional[str] = None
self.job_token: Optional[str] = None
self.oauth_token: Optional[str] = None
self.private_token: Optional[str] = None
self.api_version: str = "4"
self.order_by: Optional[str] = None
self.pagination: Optional[str] = None
self.per_page: Optional[int] = None
self.retry_transient_errors: bool = False
self.ssl_verify: Union[bool, str] = True
self.timeout: int = 60
self.url: Optional[str] = None
self.user_agent: str = USER_AGENT
self.keep_base_url: bool = False
self._files = _get_config_files(config_files)
if self._files:
self._parse_config()
if self.gitlab_id and not self._files:
raise GitlabConfigMissingError(
f"A gitlab id was provided ({self.gitlab_id}) but no config file found"
)
def _parse_config(self) -> None:
_config = configparser.ConfigParser()
_config.read(self._files, encoding="utf-8")
if self.gitlab_id and not _config.has_section(self.gitlab_id):
raise GitlabDataError(
f"A gitlab id was provided ({self.gitlab_id}) "
"but no config section found"
)
if self.gitlab_id is None:
try:
self.gitlab_id = _config.get("global", "default")
except Exception as e:
raise GitlabIDError(
"Impossible to get the gitlab id (not specified in config file)"
) from e
try:
self.url = _config.get(self.gitlab_id, "url")
except Exception as e:
raise GitlabDataError(
"Impossible to get gitlab details from "
f"configuration ({self.gitlab_id})"
) from e
try:
self.ssl_verify = _config.getboolean("global", "ssl_verify")
except ValueError:
# Value Error means the option exists but isn't a boolean.
# Get as a string instead as it should then be a local path to a
# CA bundle.
self.ssl_verify = _config.get("global", "ssl_verify")
except _CONFIG_PARSER_ERRORS:
pass
try:
self.ssl_verify = _config.getboolean(self.gitlab_id, "ssl_verify")
except ValueError:
# Value Error means the option exists but isn't a boolean.
# Get as a string instead as it should then be a local path to a
# CA bundle.
self.ssl_verify = _config.get(self.gitlab_id, "ssl_verify")
except _CONFIG_PARSER_ERRORS:
pass
try:
self.timeout = _config.getint("global", "timeout")
except _CONFIG_PARSER_ERRORS:
pass
try:
self.timeout = _config.getint(self.gitlab_id, "timeout")
except _CONFIG_PARSER_ERRORS:
pass
try:
self.private_token = _config.get(self.gitlab_id, "private_token")
except _CONFIG_PARSER_ERRORS:
pass
try:
self.oauth_token = _config.get(self.gitlab_id, "oauth_token")
except _CONFIG_PARSER_ERRORS:
pass
try:
self.job_token = _config.get(self.gitlab_id, "job_token")
except _CONFIG_PARSER_ERRORS:
pass
try:
self.http_username = _config.get(self.gitlab_id, "http_username")
self.http_password = _config.get(
self.gitlab_id, "http_password"
) # pragma: no cover
except _CONFIG_PARSER_ERRORS:
pass
self._get_values_from_helper()
try:
self.api_version = _config.get("global", "api_version")
except _CONFIG_PARSER_ERRORS:
pass
try:
self.api_version = _config.get(self.gitlab_id, "api_version")
except _CONFIG_PARSER_ERRORS:
pass
if self.api_version not in ("4",):
raise GitlabDataError(f"Unsupported API version: {self.api_version}")
for section in ["global", self.gitlab_id]:
try:
self.per_page = _config.getint(section, "per_page")
except _CONFIG_PARSER_ERRORS:
pass
if self.per_page is not None and not 0