config.py 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896
  1. # config.py
  2. # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
  3. #
  4. # This module is part of GitPython and is released under
  5. # the BSD License: http://www.opensource.org/licenses/bsd-license.php
  6. """Module containing module parser implementation able to properly read and write
  7. configuration files"""
  8. import sys
  9. import abc
  10. from functools import wraps
  11. import inspect
  12. from io import BufferedReader, IOBase
  13. import logging
  14. import os
  15. import re
  16. import fnmatch
  17. from git.compat import (
  18. defenc,
  19. force_text,
  20. is_win,
  21. )
  22. from git.util import LockFile
  23. import os.path as osp
  24. import configparser as cp
  25. # typing-------------------------------------------------------
  26. from typing import (
  27. Any,
  28. Callable,
  29. Generic,
  30. IO,
  31. List,
  32. Dict,
  33. Sequence,
  34. TYPE_CHECKING,
  35. Tuple,
  36. TypeVar,
  37. Union,
  38. cast,
  39. )
  40. from git.types import Lit_config_levels, ConfigLevels_Tup, PathLike, assert_never, _T
  41. if TYPE_CHECKING:
  42. from git.repo.base import Repo
  43. from io import BytesIO
  44. T_ConfigParser = TypeVar("T_ConfigParser", bound="GitConfigParser")
  45. T_OMD_value = TypeVar("T_OMD_value", str, bytes, int, float, bool)
  46. if sys.version_info[:3] < (3, 7, 2):
  47. # typing.Ordereddict not added until py 3.7.2
  48. from collections import OrderedDict
  49. OrderedDict_OMD = OrderedDict
  50. else:
  51. from typing import OrderedDict
  52. OrderedDict_OMD = OrderedDict[str, List[T_OMD_value]] # type: ignore[assignment, misc]
  53. # -------------------------------------------------------------
  54. __all__ = ("GitConfigParser", "SectionConstraint")
  55. log = logging.getLogger("git.config")
  56. log.addHandler(logging.NullHandler())
  57. # invariants
  58. # represents the configuration level of a configuration file
  59. CONFIG_LEVELS: ConfigLevels_Tup = ("system", "user", "global", "repository")
  60. # Section pattern to detect conditional includes.
  61. # https://git-scm.com/docs/git-config#_conditional_includes
  62. CONDITIONAL_INCLUDE_REGEXP = re.compile(r"(?<=includeIf )\"(gitdir|gitdir/i|onbranch):(.+)\"")
  63. class MetaParserBuilder(abc.ABCMeta): # noqa: B024
  64. """Utility class wrapping base-class methods into decorators that assure read-only properties"""
  65. def __new__(cls, name: str, bases: Tuple, clsdict: Dict[str, Any]) -> "MetaParserBuilder":
  66. """
  67. Equip all base-class methods with a needs_values decorator, and all non-const methods
  68. with a set_dirty_and_flush_changes decorator in addition to that."""
  69. kmm = "_mutating_methods_"
  70. if kmm in clsdict:
  71. mutating_methods = clsdict[kmm]
  72. for base in bases:
  73. methods = (t for t in inspect.getmembers(base, inspect.isroutine) if not t[0].startswith("_"))
  74. for name, method in methods:
  75. if name in clsdict:
  76. continue
  77. method_with_values = needs_values(method)
  78. if name in mutating_methods:
  79. method_with_values = set_dirty_and_flush_changes(method_with_values)
  80. # END mutating methods handling
  81. clsdict[name] = method_with_values
  82. # END for each name/method pair
  83. # END for each base
  84. # END if mutating methods configuration is set
  85. new_type = super(MetaParserBuilder, cls).__new__(cls, name, bases, clsdict)
  86. return new_type
  87. def needs_values(func: Callable[..., _T]) -> Callable[..., _T]:
  88. """Returns method assuring we read values (on demand) before we try to access them"""
  89. @wraps(func)
  90. def assure_data_present(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
  91. self.read()
  92. return func(self, *args, **kwargs)
  93. # END wrapper method
  94. return assure_data_present
  95. def set_dirty_and_flush_changes(non_const_func: Callable[..., _T]) -> Callable[..., _T]:
  96. """Return method that checks whether given non constant function may be called.
  97. If so, the instance will be set dirty.
  98. Additionally, we flush the changes right to disk"""
  99. def flush_changes(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
  100. rval = non_const_func(self, *args, **kwargs)
  101. self._dirty = True
  102. self.write()
  103. return rval
  104. # END wrapper method
  105. flush_changes.__name__ = non_const_func.__name__
  106. return flush_changes
  107. class SectionConstraint(Generic[T_ConfigParser]):
  108. """Constrains a ConfigParser to only option commands which are constrained to
  109. always use the section we have been initialized with.
  110. It supports all ConfigParser methods that operate on an option.
  111. :note:
  112. If used as a context manager, will release the wrapped ConfigParser."""
  113. __slots__ = ("_config", "_section_name")
  114. _valid_attrs_ = (
  115. "get_value",
  116. "set_value",
  117. "get",
  118. "set",
  119. "getint",
  120. "getfloat",
  121. "getboolean",
  122. "has_option",
  123. "remove_section",
  124. "remove_option",
  125. "options",
  126. )
  127. def __init__(self, config: T_ConfigParser, section: str) -> None:
  128. self._config = config
  129. self._section_name = section
  130. def __del__(self) -> None:
  131. # Yes, for some reason, we have to call it explicitly for it to work in PY3 !
  132. # Apparently __del__ doesn't get call anymore if refcount becomes 0
  133. # Ridiculous ... .
  134. self._config.release()
  135. def __getattr__(self, attr: str) -> Any:
  136. if attr in self._valid_attrs_:
  137. return lambda *args, **kwargs: self._call_config(attr, *args, **kwargs)
  138. return super(SectionConstraint, self).__getattribute__(attr)
  139. def _call_config(self, method: str, *args: Any, **kwargs: Any) -> Any:
  140. """Call the configuration at the given method which must take a section name
  141. as first argument"""
  142. return getattr(self._config, method)(self._section_name, *args, **kwargs)
  143. @property
  144. def config(self) -> T_ConfigParser:
  145. """return: Configparser instance we constrain"""
  146. return self._config
  147. def release(self) -> None:
  148. """Equivalent to GitConfigParser.release(), which is called on our underlying parser instance"""
  149. return self._config.release()
  150. def __enter__(self) -> "SectionConstraint[T_ConfigParser]":
  151. self._config.__enter__()
  152. return self
  153. def __exit__(self, exception_type: str, exception_value: str, traceback: str) -> None:
  154. self._config.__exit__(exception_type, exception_value, traceback)
  155. class _OMD(OrderedDict_OMD):
  156. """Ordered multi-dict."""
  157. def __setitem__(self, key: str, value: _T) -> None:
  158. super(_OMD, self).__setitem__(key, [value])
  159. def add(self, key: str, value: Any) -> None:
  160. if key not in self:
  161. super(_OMD, self).__setitem__(key, [value])
  162. return None
  163. super(_OMD, self).__getitem__(key).append(value)
  164. def setall(self, key: str, values: List[_T]) -> None:
  165. super(_OMD, self).__setitem__(key, values)
  166. def __getitem__(self, key: str) -> Any:
  167. return super(_OMD, self).__getitem__(key)[-1]
  168. def getlast(self, key: str) -> Any:
  169. return super(_OMD, self).__getitem__(key)[-1]
  170. def setlast(self, key: str, value: Any) -> None:
  171. if key not in self:
  172. super(_OMD, self).__setitem__(key, [value])
  173. return
  174. prior = super(_OMD, self).__getitem__(key)
  175. prior[-1] = value
  176. def get(self, key: str, default: Union[_T, None] = None) -> Union[_T, None]:
  177. return super(_OMD, self).get(key, [default])[-1]
  178. def getall(self, key: str) -> List[_T]:
  179. return super(_OMD, self).__getitem__(key)
  180. def items(self) -> List[Tuple[str, _T]]: # type: ignore[override]
  181. """List of (key, last value for key)."""
  182. return [(k, self[k]) for k in self]
  183. def items_all(self) -> List[Tuple[str, List[_T]]]:
  184. """List of (key, list of values for key)."""
  185. return [(k, self.getall(k)) for k in self]
  186. def get_config_path(config_level: Lit_config_levels) -> str:
  187. # we do not support an absolute path of the gitconfig on windows ,
  188. # use the global config instead
  189. if is_win and config_level == "system":
  190. config_level = "global"
  191. if config_level == "system":
  192. return "/etc/gitconfig"
  193. elif config_level == "user":
  194. config_home = os.environ.get("XDG_CONFIG_HOME") or osp.join(os.environ.get("HOME", "~"), ".config")
  195. return osp.normpath(osp.expanduser(osp.join(config_home, "git", "config")))
  196. elif config_level == "global":
  197. return osp.normpath(osp.expanduser("~/.gitconfig"))
  198. elif config_level == "repository":
  199. raise ValueError("No repo to get repository configuration from. Use Repo._get_config_path")
  200. else:
  201. # Should not reach here. Will raise ValueError if does. Static typing will warn missing elifs
  202. assert_never(
  203. config_level, # type: ignore[unreachable]
  204. ValueError(f"Invalid configuration level: {config_level!r}"),
  205. )
  206. class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder):
  207. """Implements specifics required to read git style configuration files.
  208. This variation behaves much like the git.config command such that the configuration
  209. will be read on demand based on the filepath given during initialization.
  210. The changes will automatically be written once the instance goes out of scope, but
  211. can be triggered manually as well.
  212. The configuration file will be locked if you intend to change values preventing other
  213. instances to write concurrently.
  214. :note:
  215. The config is case-sensitive even when queried, hence section and option names
  216. must match perfectly.
  217. If used as a context manager, will release the locked file."""
  218. # { Configuration
  219. # The lock type determines the type of lock to use in new configuration readers.
  220. # They must be compatible to the LockFile interface.
  221. # A suitable alternative would be the BlockingLockFile
  222. t_lock = LockFile
  223. re_comment = re.compile(r"^\s*[#;]")
  224. # } END configuration
  225. optvalueonly_source = r"\s*(?P<option>[^:=\s][^:=]*)"
  226. OPTVALUEONLY = re.compile(optvalueonly_source)
  227. OPTCRE = re.compile(optvalueonly_source + r"\s*(?P<vi>[:=])\s*" + r"(?P<value>.*)$")
  228. del optvalueonly_source
  229. # list of RawConfigParser methods able to change the instance
  230. _mutating_methods_ = ("add_section", "remove_section", "remove_option", "set")
  231. def __init__(
  232. self,
  233. file_or_files: Union[None, PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = None,
  234. read_only: bool = True,
  235. merge_includes: bool = True,
  236. config_level: Union[Lit_config_levels, None] = None,
  237. repo: Union["Repo", None] = None,
  238. ) -> None:
  239. """Initialize a configuration reader to read the given file_or_files and to
  240. possibly allow changes to it by setting read_only False
  241. :param file_or_files:
  242. A single file path or file objects or multiple of these
  243. :param read_only:
  244. If True, the ConfigParser may only read the data , but not change it.
  245. If False, only a single file path or file object may be given. We will write back the changes
  246. when they happen, or when the ConfigParser is released. This will not happen if other
  247. configuration files have been included
  248. :param merge_includes: if True, we will read files mentioned in [include] sections and merge their
  249. contents into ours. This makes it impossible to write back an individual configuration file.
  250. Thus, if you want to modify a single configuration file, turn this off to leave the original
  251. dataset unaltered when reading it.
  252. :param repo: Reference to repository to use if [includeIf] sections are found in configuration files.
  253. """
  254. cp.RawConfigParser.__init__(self, dict_type=_OMD)
  255. self._dict: Callable[..., _OMD] # type: ignore # mypy/typeshed bug?
  256. self._defaults: _OMD
  257. self._sections: _OMD # type: ignore # mypy/typeshed bug?
  258. # Used in python 3, needs to stay in sync with sections for underlying implementation to work
  259. if not hasattr(self, "_proxies"):
  260. self._proxies = self._dict()
  261. if file_or_files is not None:
  262. self._file_or_files: Union[PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = file_or_files
  263. else:
  264. if config_level is None:
  265. if read_only:
  266. self._file_or_files = [
  267. get_config_path(cast(Lit_config_levels, f)) for f in CONFIG_LEVELS if f != "repository"
  268. ]
  269. else:
  270. raise ValueError("No configuration level or configuration files specified")
  271. else:
  272. self._file_or_files = [get_config_path(config_level)]
  273. self._read_only = read_only
  274. self._dirty = False
  275. self._is_initialized = False
  276. self._merge_includes = merge_includes
  277. self._repo = repo
  278. self._lock: Union["LockFile", None] = None
  279. self._acquire_lock()
  280. def _acquire_lock(self) -> None:
  281. if not self._read_only:
  282. if not self._lock:
  283. if isinstance(self._file_or_files, (str, os.PathLike)):
  284. file_or_files = self._file_or_files
  285. elif isinstance(self._file_or_files, (tuple, list, Sequence)):
  286. raise ValueError(
  287. "Write-ConfigParsers can operate on a single file only, multiple files have been passed"
  288. )
  289. else:
  290. file_or_files = self._file_or_files.name
  291. # END get filename from handle/stream
  292. # initialize lock base - we want to write
  293. self._lock = self.t_lock(file_or_files)
  294. # END lock check
  295. self._lock._obtain_lock()
  296. # END read-only check
  297. def __del__(self) -> None:
  298. """Write pending changes if required and release locks"""
  299. # NOTE: only consistent in PY2
  300. self.release()
  301. def __enter__(self) -> "GitConfigParser":
  302. self._acquire_lock()
  303. return self
  304. def __exit__(self, *args: Any) -> None:
  305. self.release()
  306. def release(self) -> None:
  307. """Flush changes and release the configuration write lock. This instance must not be used anymore afterwards.
  308. In Python 3, it's required to explicitly release locks and flush changes, as __del__ is not called
  309. deterministically anymore."""
  310. # checking for the lock here makes sure we do not raise during write()
  311. # in case an invalid parser was created who could not get a lock
  312. if self.read_only or (self._lock and not self._lock._has_lock()):
  313. return
  314. try:
  315. try:
  316. self.write()
  317. except IOError:
  318. log.error("Exception during destruction of GitConfigParser", exc_info=True)
  319. except ReferenceError:
  320. # This happens in PY3 ... and usually means that some state cannot be written
  321. # as the sections dict cannot be iterated
  322. # Usually when shutting down the interpreter, don'y know how to fix this
  323. pass
  324. finally:
  325. if self._lock is not None:
  326. self._lock._release_lock()
  327. def optionxform(self, optionstr: str) -> str:
  328. """Do not transform options in any way when writing"""
  329. return optionstr
  330. def _read(self, fp: Union[BufferedReader, IO[bytes]], fpname: str) -> None:
  331. """A direct copy of the py2.4 version of the super class's _read method
  332. to assure it uses ordered dicts. Had to change one line to make it work.
  333. Future versions have this fixed, but in fact its quite embarrassing for the
  334. guys not to have done it right in the first place !
  335. Removed big comments to make it more compact.
  336. Made sure it ignores initial whitespace as git uses tabs"""
  337. cursect = None # None, or a dictionary
  338. optname = None
  339. lineno = 0
  340. is_multi_line = False
  341. e = None # None, or an exception
  342. def string_decode(v: str) -> str:
  343. if v[-1] == "\\":
  344. v = v[:-1]
  345. # end cut trailing escapes to prevent decode error
  346. return v.encode(defenc).decode("unicode_escape")
  347. # end
  348. # end
  349. while True:
  350. # we assume to read binary !
  351. line = fp.readline().decode(defenc)
  352. if not line:
  353. break
  354. lineno = lineno + 1
  355. # comment or blank line?
  356. if line.strip() == "" or self.re_comment.match(line):
  357. continue
  358. if line.split(None, 1)[0].lower() == "rem" and line[0] in "rR":
  359. # no leading whitespace
  360. continue
  361. # is it a section header?
  362. mo = self.SECTCRE.match(line.strip())
  363. if not is_multi_line and mo:
  364. sectname: str = mo.group("header").strip()
  365. if sectname in self._sections:
  366. cursect = self._sections[sectname]
  367. elif sectname == cp.DEFAULTSECT:
  368. cursect = self._defaults
  369. else:
  370. cursect = self._dict((("__name__", sectname),))
  371. self._sections[sectname] = cursect
  372. self._proxies[sectname] = None
  373. # So sections can't start with a continuation line
  374. optname = None
  375. # no section header in the file?
  376. elif cursect is None:
  377. raise cp.MissingSectionHeaderError(fpname, lineno, line)
  378. # an option line?
  379. elif not is_multi_line:
  380. mo = self.OPTCRE.match(line)
  381. if mo:
  382. # We might just have handled the last line, which could contain a quotation we want to remove
  383. optname, vi, optval = mo.group("option", "vi", "value")
  384. if vi in ("=", ":") and ";" in optval and not optval.strip().startswith('"'):
  385. pos = optval.find(";")
  386. if pos != -1 and optval[pos - 1].isspace():
  387. optval = optval[:pos]
  388. optval = optval.strip()
  389. if optval == '""':
  390. optval = ""
  391. # end handle empty string
  392. optname = self.optionxform(optname.rstrip())
  393. if len(optval) > 1 and optval[0] == '"' and optval[-1] != '"':
  394. is_multi_line = True
  395. optval = string_decode(optval[1:])
  396. # end handle multi-line
  397. # preserves multiple values for duplicate optnames
  398. cursect.add(optname, optval)
  399. else:
  400. # check if it's an option with no value - it's just ignored by git
  401. if not self.OPTVALUEONLY.match(line):
  402. if not e:
  403. e = cp.ParsingError(fpname)
  404. e.append(lineno, repr(line))
  405. continue
  406. else:
  407. line = line.rstrip()
  408. if line.endswith('"'):
  409. is_multi_line = False
  410. line = line[:-1]
  411. # end handle quotations
  412. optval = cursect.getlast(optname)
  413. cursect.setlast(optname, optval + string_decode(line))
  414. # END parse section or option
  415. # END while reading
  416. # if any parsing errors occurred, raise an exception
  417. if e:
  418. raise e
  419. def _has_includes(self) -> Union[bool, int]:
  420. return self._merge_includes and len(self._included_paths())
  421. def _included_paths(self) -> List[Tuple[str, str]]:
  422. """Return List all paths that must be included to configuration
  423. as Tuples of (option, value).
  424. """
  425. paths = []
  426. for section in self.sections():
  427. if section == "include":
  428. paths += self.items(section)
  429. match = CONDITIONAL_INCLUDE_REGEXP.search(section)
  430. if match is None or self._repo is None:
  431. continue
  432. keyword = match.group(1)
  433. value = match.group(2).strip()
  434. if keyword in ["gitdir", "gitdir/i"]:
  435. value = osp.expanduser(value)
  436. if not any(value.startswith(s) for s in ["./", "/"]):
  437. value = "**/" + value
  438. if value.endswith("/"):
  439. value += "**"
  440. # Ensure that glob is always case insensitive if required.
  441. if keyword.endswith("/i"):
  442. value = re.sub(
  443. r"[a-zA-Z]",
  444. lambda m: "[{}{}]".format(m.group().lower(), m.group().upper()),
  445. value,
  446. )
  447. if self._repo.git_dir:
  448. if fnmatch.fnmatchcase(str(self._repo.git_dir), value):
  449. paths += self.items(section)
  450. elif keyword == "onbranch":
  451. try:
  452. branch_name = self._repo.active_branch.name
  453. except TypeError:
  454. # Ignore section if active branch cannot be retrieved.
  455. continue
  456. if fnmatch.fnmatchcase(branch_name, value):
  457. paths += self.items(section)
  458. return paths
  459. def read(self) -> None: # type: ignore[override]
  460. """Reads the data stored in the files we have been initialized with. It will
  461. ignore files that cannot be read, possibly leaving an empty configuration
  462. :return: Nothing
  463. :raise IOError: if a file cannot be handled"""
  464. if self._is_initialized:
  465. return None
  466. self._is_initialized = True
  467. files_to_read: List[Union[PathLike, IO]] = [""]
  468. if isinstance(self._file_or_files, (str, os.PathLike)):
  469. # for str or Path, as str is a type of Sequence
  470. files_to_read = [self._file_or_files]
  471. elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
  472. # could merge with above isinstance once runtime type known
  473. files_to_read = [self._file_or_files]
  474. else: # for lists or tuples
  475. files_to_read = list(self._file_or_files)
  476. # end assure we have a copy of the paths to handle
  477. seen = set(files_to_read)
  478. num_read_include_files = 0
  479. while files_to_read:
  480. file_path = files_to_read.pop(0)
  481. file_ok = False
  482. if hasattr(file_path, "seek"):
  483. # must be a file objectfile-object
  484. file_path = cast(IO[bytes], file_path) # replace with assert to narrow type, once sure
  485. self._read(file_path, file_path.name)
  486. else:
  487. # assume a path if it is not a file-object
  488. file_path = cast(PathLike, file_path)
  489. try:
  490. with open(file_path, "rb") as fp:
  491. file_ok = True
  492. self._read(fp, fp.name)
  493. except IOError:
  494. continue
  495. # Read includes and append those that we didn't handle yet
  496. # We expect all paths to be normalized and absolute (and will assure that is the case)
  497. if self._has_includes():
  498. for _, include_path in self._included_paths():
  499. if include_path.startswith("~"):
  500. include_path = osp.expanduser(include_path)
  501. if not osp.isabs(include_path):
  502. if not file_ok:
  503. continue
  504. # end ignore relative paths if we don't know the configuration file path
  505. file_path = cast(PathLike, file_path)
  506. assert osp.isabs(file_path), "Need absolute paths to be sure our cycle checks will work"
  507. include_path = osp.join(osp.dirname(file_path), include_path)
  508. # end make include path absolute
  509. include_path = osp.normpath(include_path)
  510. if include_path in seen or not os.access(include_path, os.R_OK):
  511. continue
  512. seen.add(include_path)
  513. # insert included file to the top to be considered first
  514. files_to_read.insert(0, include_path)
  515. num_read_include_files += 1
  516. # each include path in configuration file
  517. # end handle includes
  518. # END for each file object to read
  519. # If there was no file included, we can safely write back (potentially) the configuration file
  520. # without altering it's meaning
  521. if num_read_include_files == 0:
  522. self._merge_includes = False
  523. # end
  524. def _write(self, fp: IO) -> None:
  525. """Write an .ini-format representation of the configuration state in
  526. git compatible format"""
  527. def write_section(name: str, section_dict: _OMD) -> None:
  528. fp.write(("[%s]\n" % name).encode(defenc))
  529. values: Sequence[str] # runtime only gets str in tests, but should be whatever _OMD stores
  530. v: str
  531. for (key, values) in section_dict.items_all():
  532. if key == "__name__":
  533. continue
  534. for v in values:
  535. fp.write(("\t%s = %s\n" % (key, self._value_to_string(v).replace("\n", "\n\t"))).encode(defenc))
  536. # END if key is not __name__
  537. # END section writing
  538. if self._defaults:
  539. write_section(cp.DEFAULTSECT, self._defaults)
  540. value: _OMD
  541. for name, value in self._sections.items():
  542. write_section(name, value)
  543. def items(self, section_name: str) -> List[Tuple[str, str]]: # type: ignore[override]
  544. """:return: list((option, value), ...) pairs of all items in the given section"""
  545. return [(k, v) for k, v in super(GitConfigParser, self).items(section_name) if k != "__name__"]
  546. def items_all(self, section_name: str) -> List[Tuple[str, List[str]]]:
  547. """:return: list((option, [values...]), ...) pairs of all items in the given section"""
  548. rv = _OMD(self._defaults)
  549. for k, vs in self._sections[section_name].items_all():
  550. if k == "__name__":
  551. continue
  552. if k in rv and rv.getall(k) == vs:
  553. continue
  554. for v in vs:
  555. rv.add(k, v)
  556. return rv.items_all()
  557. @needs_values
  558. def write(self) -> None:
  559. """Write changes to our file, if there are changes at all
  560. :raise IOError: if this is a read-only writer instance or if we could not obtain
  561. a file lock"""
  562. self._assure_writable("write")
  563. if not self._dirty:
  564. return None
  565. if isinstance(self._file_or_files, (list, tuple)):
  566. raise AssertionError(
  567. "Cannot write back if there is not exactly a single file to write to, have %i files"
  568. % len(self._file_or_files)
  569. )
  570. # end assert multiple files
  571. if self._has_includes():
  572. log.debug(
  573. "Skipping write-back of configuration file as include files were merged in."
  574. + "Set merge_includes=False to prevent this."
  575. )
  576. return None
  577. # end
  578. fp = self._file_or_files
  579. # we have a physical file on disk, so get a lock
  580. is_file_lock = isinstance(fp, (str, os.PathLike, IOBase)) # can't use Pathlike until 3.5 dropped
  581. if is_file_lock and self._lock is not None: # else raise Error?
  582. self._lock._obtain_lock()
  583. if not hasattr(fp, "seek"):
  584. fp = cast(PathLike, fp)
  585. with open(fp, "wb") as fp_open:
  586. self._write(fp_open)
  587. else:
  588. fp = cast("BytesIO", fp)
  589. fp.seek(0)
  590. # make sure we do not overwrite into an existing file
  591. if hasattr(fp, "truncate"):
  592. fp.truncate()
  593. self._write(fp)
  594. def _assure_writable(self, method_name: str) -> None:
  595. if self.read_only:
  596. raise IOError("Cannot execute non-constant method %s.%s" % (self, method_name))
  597. def add_section(self, section: str) -> None:
  598. """Assures added options will stay in order"""
  599. return super(GitConfigParser, self).add_section(section)
  600. @property
  601. def read_only(self) -> bool:
  602. """:return: True if this instance may change the configuration file"""
  603. return self._read_only
  604. def get_value(
  605. self,
  606. section: str,
  607. option: str,
  608. default: Union[int, float, str, bool, None] = None,
  609. ) -> Union[int, float, str, bool]:
  610. # can default or return type include bool?
  611. """Get an option's value.
  612. If multiple values are specified for this option in the section, the
  613. last one specified is returned.
  614. :param default:
  615. If not None, the given default value will be returned in case
  616. the option did not exist
  617. :return: a properly typed value, either int, float or string
  618. :raise TypeError: in case the value could not be understood
  619. Otherwise the exceptions known to the ConfigParser will be raised."""
  620. try:
  621. valuestr = self.get(section, option)
  622. except Exception:
  623. if default is not None:
  624. return default
  625. raise
  626. return self._string_to_value(valuestr)
  627. def get_values(
  628. self,
  629. section: str,
  630. option: str,
  631. default: Union[int, float, str, bool, None] = None,
  632. ) -> List[Union[int, float, str, bool]]:
  633. """Get an option's values.
  634. If multiple values are specified for this option in the section, all are
  635. returned.
  636. :param default:
  637. If not None, a list containing the given default value will be
  638. returned in case the option did not exist
  639. :return: a list of properly typed values, either int, float or string
  640. :raise TypeError: in case the value could not be understood
  641. Otherwise the exceptions known to the ConfigParser will be raised."""
  642. try:
  643. lst = self._sections[section].getall(option)
  644. except Exception:
  645. if default is not None:
  646. return [default]
  647. raise
  648. return [self._string_to_value(valuestr) for valuestr in lst]
  649. def _string_to_value(self, valuestr: str) -> Union[int, float, str, bool]:
  650. types = (int, float)
  651. for numtype in types:
  652. try:
  653. val = numtype(valuestr)
  654. # truncated value ?
  655. if val != float(valuestr):
  656. continue
  657. return val
  658. except (ValueError, TypeError):
  659. continue
  660. # END for each numeric type
  661. # try boolean values as git uses them
  662. vl = valuestr.lower()
  663. if vl == "false":
  664. return False
  665. if vl == "true":
  666. return True
  667. if not isinstance(valuestr, str):
  668. raise TypeError(
  669. "Invalid value type: only int, long, float and str are allowed",
  670. valuestr,
  671. )
  672. return valuestr
  673. def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str:
  674. if isinstance(value, (int, float, bool)):
  675. return str(value)
  676. return force_text(value)
  677. @needs_values
  678. @set_dirty_and_flush_changes
  679. def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
  680. """Sets the given option in section to the given value.
  681. It will create the section if required, and will not throw as opposed to the default
  682. ConfigParser 'set' method.
  683. :param section: Name of the section in which the option resides or should reside
  684. :param option: Name of the options whose value to set
  685. :param value: Value to set the option to. It must be a string or convertible
  686. to a string
  687. :return: this instance"""
  688. if not self.has_section(section):
  689. self.add_section(section)
  690. self.set(section, option, self._value_to_string(value))
  691. return self
  692. @needs_values
  693. @set_dirty_and_flush_changes
  694. def add_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
  695. """Adds a value for the given option in section.
  696. It will create the section if required, and will not throw as opposed to the default
  697. ConfigParser 'set' method. The value becomes the new value of the option as returned
  698. by 'get_value', and appends to the list of values returned by 'get_values`'.
  699. :param section: Name of the section in which the option resides or should reside
  700. :param option: Name of the option
  701. :param value: Value to add to option. It must be a string or convertible
  702. to a string
  703. :return: this instance"""
  704. if not self.has_section(section):
  705. self.add_section(section)
  706. self._sections[section].add(option, self._value_to_string(value))
  707. return self
  708. def rename_section(self, section: str, new_name: str) -> "GitConfigParser":
  709. """rename the given section to new_name
  710. :raise ValueError: if section doesn't exit
  711. :raise ValueError: if a section with new_name does already exist
  712. :return: this instance
  713. """
  714. if not self.has_section(section):
  715. raise ValueError("Source section '%s' doesn't exist" % section)
  716. if self.has_section(new_name):
  717. raise ValueError("Destination section '%s' already exists" % new_name)
  718. super(GitConfigParser, self).add_section(new_name)
  719. new_section = self._sections[new_name]
  720. for k, vs in self.items_all(section):
  721. new_section.setall(k, vs)
  722. # end for each value to copy
  723. # This call writes back the changes, which is why we don't have the respective decorator
  724. self.remove_section(section)
  725. return self