Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/config.py: 74%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
2#
3# This module is part of GitPython and is released under the
4# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
6"""Parser for reading and writing configuration files."""
8__all__ = ["GitConfigParser", "SectionConstraint"]
10import abc
11import configparser as cp
12import fnmatch
13from functools import wraps
14import inspect
15from io import BufferedReader, IOBase
16import logging
17import os
18import os.path as osp
19import re
20import sys
22from git.compat import defenc, force_text
23from git.util import LockFile
25# typing-------------------------------------------------------
27from typing import (
28 Any,
29 Callable,
30 Generic,
31 IO,
32 List,
33 Dict,
34 Sequence,
35 TYPE_CHECKING,
36 Tuple,
37 TypeVar,
38 Union,
39 cast,
40)
42from git.types import Lit_config_levels, ConfigLevels_Tup, PathLike, assert_never, _T
44if TYPE_CHECKING:
45 from io import BytesIO
47 from git.repo.base import Repo
49T_ConfigParser = TypeVar("T_ConfigParser", bound="GitConfigParser")
50T_OMD_value = TypeVar("T_OMD_value", str, bytes, int, float, bool)
52if sys.version_info[:3] < (3, 7, 2):
53 # typing.Ordereddict not added until Python 3.7.2.
54 from collections import OrderedDict
56 OrderedDict_OMD = OrderedDict
57else:
58 from typing import OrderedDict
60 OrderedDict_OMD = OrderedDict[str, List[T_OMD_value]] # type: ignore[assignment, misc]
62# -------------------------------------------------------------
64_logger = logging.getLogger(__name__)
66CONFIG_LEVELS: ConfigLevels_Tup = ("system", "user", "global", "repository")
67"""The configuration level of a configuration file."""
69CONDITIONAL_INCLUDE_REGEXP = re.compile(r"(?<=includeIf )\"(gitdir|gitdir/i|onbranch|hasconfig:remote\.\*\.url):(.+)\"")
70"""Section pattern to detect conditional includes.
72See: https://git-scm.com/docs/git-config#_conditional_includes
73"""
75UNSAFE_CONFIG_CHARS_RE = re.compile(r"[\r\n\x00]")
76"""Characters that cannot be safely written in config names or values."""
78VALID_CONFIG_OPTION_NAME_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
79"""Pattern for option names that can be written without changing config syntax."""
82class MetaParserBuilder(abc.ABCMeta): # noqa: B024
83 """Utility class wrapping base-class methods into decorators that assure read-only
84 properties."""
86 def __new__(cls, name: str, bases: Tuple, clsdict: Dict[str, Any]) -> "MetaParserBuilder":
87 """Equip all base-class methods with a needs_values decorator, and all non-const
88 methods with a :func:`set_dirty_and_flush_changes` decorator in addition to
89 that.
90 """
91 kmm = "_mutating_methods_"
92 if kmm in clsdict:
93 mutating_methods = clsdict[kmm]
94 for base in bases:
95 methods = (t for t in inspect.getmembers(base, inspect.isroutine) if not t[0].startswith("_"))
96 for method_name, method in methods:
97 if method_name in clsdict:
98 continue
99 method_with_values = needs_values(method)
100 if method_name in mutating_methods:
101 method_with_values = set_dirty_and_flush_changes(method_with_values)
102 # END mutating methods handling
104 clsdict[method_name] = method_with_values
105 # END for each name/method pair
106 # END for each base
107 # END if mutating methods configuration is set
109 new_type = super().__new__(cls, name, bases, clsdict)
110 return new_type
113def needs_values(func: Callable[..., _T]) -> Callable[..., _T]:
114 """Return a method for ensuring we read values (on demand) before we try to access
115 them."""
117 @wraps(func)
118 def assure_data_present(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
119 self.read()
120 return func(self, *args, **kwargs)
122 # END wrapper method
123 return assure_data_present
126def set_dirty_and_flush_changes(non_const_func: Callable[..., _T]) -> Callable[..., _T]:
127 """Return a method that checks whether given non constant function may be called.
129 If so, the instance will be set dirty. Additionally, we flush the changes right to
130 disk.
131 """
133 def flush_changes(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
134 rval = non_const_func(self, *args, **kwargs)
135 self._dirty = True
136 self.write()
137 return rval
139 # END wrapper method
140 flush_changes.__name__ = non_const_func.__name__
141 return flush_changes
144class SectionConstraint(Generic[T_ConfigParser]):
145 """Constrains a ConfigParser to only option commands which are constrained to
146 always use the section we have been initialized with.
148 It supports all ConfigParser methods that operate on an option.
150 :note:
151 If used as a context manager, will release the wrapped ConfigParser.
152 """
154 __slots__ = ("_config", "_section_name")
156 _valid_attrs_ = (
157 "get_value",
158 "set_value",
159 "get",
160 "set",
161 "getint",
162 "getfloat",
163 "getboolean",
164 "has_option",
165 "remove_section",
166 "remove_option",
167 "options",
168 )
170 def __init__(self, config: T_ConfigParser, section: str) -> None:
171 self._config = config
172 self._section_name = section
174 def __del__(self) -> None:
175 # Yes, for some reason, we have to call it explicitly for it to work in PY3 !
176 # Apparently __del__ doesn't get call anymore if refcount becomes 0
177 # Ridiculous ... .
178 self._config.release()
180 def __getattr__(self, attr: str) -> Any:
181 if attr in self._valid_attrs_:
182 return lambda *args, **kwargs: self._call_config(attr, *args, **kwargs)
183 return super().__getattribute__(attr)
185 def _call_config(self, method: str, *args: Any, **kwargs: Any) -> Any:
186 """Call the configuration at the given method which must take a section name as
187 first argument."""
188 return getattr(self._config, method)(self._section_name, *args, **kwargs)
190 @property
191 def config(self) -> T_ConfigParser:
192 """return: ConfigParser instance we constrain"""
193 return self._config
195 def release(self) -> None:
196 """Equivalent to :meth:`GitConfigParser.release`, which is called on our
197 underlying parser instance."""
198 return self._config.release()
200 def __enter__(self) -> "SectionConstraint[T_ConfigParser]":
201 self._config.__enter__()
202 return self
204 def __exit__(self, exception_type: str, exception_value: str, traceback: str) -> None:
205 self._config.__exit__(exception_type, exception_value, traceback)
208class _OMD(OrderedDict_OMD):
209 """Ordered multi-dict."""
211 def __setitem__(self, key: str, value: _T) -> None:
212 super().__setitem__(key, [value])
214 def add(self, key: str, value: Any) -> None:
215 if key not in self:
216 super().__setitem__(key, [value])
217 return
219 super().__getitem__(key).append(value)
221 def setall(self, key: str, values: List[_T]) -> None:
222 super().__setitem__(key, values)
224 def __getitem__(self, key: str) -> Any:
225 return super().__getitem__(key)[-1]
227 def getlast(self, key: str) -> Any:
228 return super().__getitem__(key)[-1]
230 def setlast(self, key: str, value: Any) -> None:
231 if key not in self:
232 super().__setitem__(key, [value])
233 return
235 prior = super().__getitem__(key)
236 prior[-1] = value
238 def get(self, key: str, default: Union[_T, None] = None) -> Union[_T, None]:
239 return super().get(key, [default])[-1]
241 def getall(self, key: str) -> List[_T]:
242 return super().__getitem__(key)
244 def items(self) -> List[Tuple[str, _T]]: # type: ignore[override]
245 """List of (key, last value for key)."""
246 return [(k, self[k]) for k in self]
248 def items_all(self) -> List[Tuple[str, List[_T]]]:
249 """List of (key, list of values for key)."""
250 return [(k, self.getall(k)) for k in self]
253def get_config_path(config_level: Lit_config_levels) -> str:
254 # We do not support an absolute path of the gitconfig on Windows.
255 # Use the global config instead.
256 if sys.platform == "win32" and config_level == "system":
257 config_level = "global"
259 if config_level == "system":
260 return "/etc/gitconfig"
261 elif config_level == "user":
262 config_home = os.environ.get("XDG_CONFIG_HOME") or osp.join(os.environ.get("HOME", "~"), ".config")
263 return osp.normpath(osp.expanduser(osp.join(config_home, "git", "config")))
264 elif config_level == "global":
265 return osp.normpath(osp.expanduser("~/.gitconfig"))
266 elif config_level == "repository":
267 raise ValueError("No repo to get repository configuration from. Use Repo._get_config_path")
268 else:
269 # Should not reach here. Will raise ValueError if does. Static typing will warn
270 # about missing elifs.
271 assert_never( # type: ignore[unreachable]
272 config_level,
273 ValueError(f"Invalid configuration level: {config_level!r}"),
274 )
277class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder):
278 """Implements specifics required to read git style configuration files.
280 This variation behaves much like the :manpage:`git-config(1)` command, such that the
281 configuration will be read on demand based on the filepath given during
282 initialization.
284 The changes will automatically be written once the instance goes out of scope, but
285 can be triggered manually as well.
287 The configuration file will be locked if you intend to change values preventing
288 other instances to write concurrently.
290 :note:
291 The config is case-sensitive even when queried, hence section and option names
292 must match perfectly.
294 :note:
295 If used as a context manager, this will release the locked file.
296 """
298 # { Configuration
299 t_lock = LockFile
300 """The lock type determines the type of lock to use in new configuration readers.
302 They must be compatible to the :class:`~git.util.LockFile` interface.
303 A suitable alternative would be the :class:`~git.util.BlockingLockFile`.
304 """
306 re_comment = re.compile(r"^\s*[#;]")
307 # } END configuration
309 optvalueonly_source = r"\s*(?P<option>[^:=\s][^:=]*)"
311 OPTVALUEONLY = re.compile(optvalueonly_source)
313 OPTCRE = re.compile(optvalueonly_source + r"\s*(?P<vi>[:=])\s*" + r"(?P<value>.*)$")
315 del optvalueonly_source
317 _mutating_methods_ = ("add_section", "remove_section", "remove_option", "set")
318 """Names of :class:`~configparser.RawConfigParser` methods able to change the
319 instance."""
321 def __init__(
322 self,
323 file_or_files: Union[None, PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = None,
324 read_only: bool = True,
325 merge_includes: bool = True,
326 config_level: Union[Lit_config_levels, None] = None,
327 repo: Union["Repo", None] = None,
328 ) -> None:
329 """Initialize a configuration reader to read the given `file_or_files` and to
330 possibly allow changes to it by setting `read_only` False.
332 :param file_or_files:
333 A file path or file object, or a sequence of possibly more than one of them.
335 :param read_only:
336 If ``True``, the ConfigParser may only read the data, but not change it.
337 If ``False``, only a single file path or file object may be given. We will
338 write back the changes when they happen, or when the ConfigParser is
339 released. This will not happen if other configuration files have been
340 included.
342 :param merge_includes:
343 If ``True``, we will read files mentioned in ``[include]`` sections and
344 merge their contents into ours. This makes it impossible to write back an
345 individual configuration file. Thus, if you want to modify a single
346 configuration file, turn this off to leave the original dataset unaltered
347 when reading it.
349 :param repo:
350 Reference to repository to use if ``[includeIf]`` sections are found in
351 configuration files.
352 """
353 cp.RawConfigParser.__init__(self, dict_type=_OMD)
354 self._dict: Callable[..., _OMD]
355 self._defaults: _OMD
356 self._sections: _OMD
358 # Used in Python 3. Needs to stay in sync with sections for underlying
359 # implementation to work.
360 if not hasattr(self, "_proxies"):
361 self._proxies = self._dict()
363 if file_or_files is not None:
364 self._file_or_files: Union[PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = file_or_files
365 else:
366 if config_level is None:
367 if read_only:
368 self._file_or_files = [
369 get_config_path(cast(Lit_config_levels, f)) for f in CONFIG_LEVELS if f != "repository"
370 ]
371 else:
372 raise ValueError("No configuration level or configuration files specified")
373 else:
374 self._file_or_files = [get_config_path(config_level)]
376 self._read_only = read_only
377 self._dirty = False
378 self._is_initialized = False
379 self._merge_includes = merge_includes
380 self._repo = repo
381 self._lock: Union["LockFile", None] = None
382 self._acquire_lock()
384 def _acquire_lock(self) -> None:
385 if not self._read_only:
386 if not self._lock:
387 if isinstance(self._file_or_files, (str, os.PathLike)):
388 file_or_files = self._file_or_files
389 elif isinstance(self._file_or_files, (tuple, list, Sequence)):
390 raise ValueError(
391 "Write-ConfigParsers can operate on a single file only, multiple files have been passed"
392 )
393 else:
394 file_or_files = self._file_or_files.name
396 # END get filename from handle/stream
397 # Initialize lock base - we want to write.
398 self._lock = self.t_lock(file_or_files)
399 # END lock check
401 self._lock._obtain_lock()
402 # END read-only check
404 def __del__(self) -> None:
405 """Write pending changes if required and release locks."""
406 # NOTE: Only consistent in Python 2.
407 self.release()
409 def __enter__(self) -> "GitConfigParser":
410 self._acquire_lock()
411 return self
413 def __exit__(self, *args: Any) -> None:
414 self.release()
416 def release(self) -> None:
417 """Flush changes and release the configuration write lock. This instance must
418 not be used anymore afterwards.
420 In Python 3, it's required to explicitly release locks and flush changes, as
421 ``__del__`` is not called deterministically anymore.
422 """
423 # Checking for the lock here makes sure we do not raise during write()
424 # in case an invalid parser was created who could not get a lock.
425 if self.read_only or (self._lock and not self._lock._has_lock()):
426 return
428 try:
429 self.write()
430 except IOError:
431 _logger.error("Exception during destruction of GitConfigParser", exc_info=True)
432 except ReferenceError:
433 # This happens in Python 3... and usually means that some state cannot be
434 # written as the sections dict cannot be iterated. This usually happens when
435 # the interpreter is shutting down. Can it be fixed?
436 pass
437 finally:
438 if self._lock is not None:
439 self._lock._release_lock()
441 def optionxform(self, optionstr: str) -> str:
442 """Do not transform options in any way when writing."""
443 return optionstr
445 def _read(self, fp: Union[BufferedReader, IO[bytes]], fpname: str) -> None:
446 """Originally a direct copy of the Python 2.4 version of
447 :meth:`RawConfigParser._read <configparser.RawConfigParser._read>`, to ensure it
448 uses ordered dicts.
450 The ordering bug was fixed in Python 2.4, and dict itself keeps ordering since
451 Python 3.7. This has some other changes, especially that it ignores initial
452 whitespace, since git uses tabs. (Big comments are removed to be more compact.)
453 """
454 cursect = None # None, or a dictionary.
455 optname = None
456 lineno = 0
457 is_multi_line = False
458 e = None # None, or an exception.
460 def string_decode(v: str) -> str:
461 if v and v.endswith("\\"):
462 v = v[:-1]
463 # END cut trailing escapes to prevent decode error
465 escapes = {"b": "\b", "n": "\n", "r": "\r", "t": "\t", '"': '"', "\\": "\\"}
466 return re.sub(r"\\(.)", lambda match: escapes.get(match.group(1), match.group(0)), v)
468 # END string_decode
470 def is_line_continuation(value: str) -> bool:
471 quoted = escaped = False
472 for char in value:
473 if escaped:
474 escaped = False
475 elif char == "\\":
476 escaped = True
477 elif char == '"':
478 quoted = not quoted
479 elif char in "#;" and not quoted:
480 return False
481 return escaped
483 def parse_value(value: str) -> str:
484 parsed: List[str] = []
485 whitespace: List[str] = []
486 quoted = escaped = False
487 escapes = {"b": "\b", "n": "\n", "t": "\t", '"': '"', "\\": "\\"}
488 for char in value:
489 if escaped:
490 parsed.append(escapes.get(char, "\\" + char))
491 escaped = False
492 continue
493 if char.isspace() and not quoted:
494 if parsed:
495 whitespace.append(char)
496 continue
497 if char in "#;" and not quoted:
498 break
499 parsed.extend(whitespace)
500 whitespace.clear()
501 if char == "\\":
502 escaped = True
503 elif char == '"':
504 quoted = not quoted
505 else:
506 parsed.append(char)
507 return "".join(parsed)
509 while True:
510 # We assume to read binary!
511 line = fp.readline().decode(defenc)
512 if not line:
513 break
514 lineno = lineno + 1
515 # Comment or blank line?
516 if line.strip() == "" or self.re_comment.match(line):
517 continue
518 if line.split(None, 1)[0].lower() == "rem" and line[0] in "rR":
519 # No leading whitespace.
520 continue
522 # Is it a section header?
523 mo = self.SECTCRE.match(line.strip())
524 if not is_multi_line and mo:
525 sectname: str = mo.group("header").strip()
526 if sectname in self._sections:
527 cursect = self._sections[sectname]
528 elif sectname == cp.DEFAULTSECT:
529 cursect = self._defaults
530 else:
531 cursect = self._dict((("__name__", sectname),))
532 self._sections[sectname] = cursect
533 self._proxies[sectname] = None
534 # So sections can't start with a continuation line.
535 optname = None
536 # No section header in the file?
537 elif cursect is None:
538 raise cp.MissingSectionHeaderError(fpname, lineno, line)
539 # An option line?
540 elif not is_multi_line:
541 mo = self.OPTCRE.match(line)
542 if mo:
543 # We might just have handled the last line, which could contain a quotation we want to remove.
544 optname, vi, optval = mo.group("option", "vi", "value")
545 optname = self.optionxform(optname.rstrip())
547 if vi in ("=", ":") and ";" in optval and not optval.strip().startswith('"'):
548 pos = optval.find(";")
549 if pos != -1 and optval[pos - 1].isspace():
550 optval = optval[:pos]
551 optval = optval.strip()
553 if len(optval) < 2 or optval[0] != '"':
554 # Does not open quoting.
555 # A value ending in an odd number of backslashes
556 # continues on the next line, exactly as git does: the
557 # final backslash and the newline are removed and the
558 # next line is appended before the complete value is
559 # parsed. An even number means the last backslash is
560 # escaped and the value ends there.
561 continued = False
562 while True:
563 if not is_line_continuation(optval):
564 break
565 continuation = fp.readline()
566 if not continuation:
567 # Backslash at end of file: git drops it.
568 optval = optval[:-1]
569 break
570 lineno = lineno + 1
571 joined = continuation.decode(defenc)
572 while joined.endswith("\n") or joined.endswith("\r"):
573 joined = joined[:-1]
574 optval = optval[:-1] + joined
575 continued = True
576 if continued:
577 optval = parse_value(optval)
578 elif optval[-1] != '"':
579 # Opens quoting and does not close: appears to start multi-line quoting.
580 is_multi_line = True
581 optval = string_decode(optval[1:])
582 elif re.search(r'(?:^|[^\\])(?:\\\\)*"', optval[1:-1]):
583 # Preserve malformed values containing unescaped quotes.
584 pass
585 else:
586 # Opens and closes quoting.
587 optval = string_decode(optval[1:-1])
589 # Preserves multiple values for duplicate optnames.
590 cursect.add(optname, optval)
591 else:
592 # Check if it's an option with no value - it's just ignored by git.
593 if not self.OPTVALUEONLY.match(line):
594 if not e:
595 e = cp.ParsingError(fpname)
596 e.append(lineno, repr(line))
597 continue
598 else:
599 line = line.rstrip()
600 if line.endswith('"'):
601 is_multi_line = False
602 line = line[:-1]
603 # END handle quotations
604 optval = cursect.getlast(optname)
605 cursect.setlast(optname, optval + string_decode(line))
606 # END parse section or option
607 # END while reading
609 # If any parsing errors occurred, raise an exception.
610 if e:
611 raise e
613 def _has_includes(self) -> Union[bool, int]:
614 return self._merge_includes and len(self._included_paths())
616 def _included_paths(self) -> List[Tuple[str, str]]:
617 """List all paths that must be included to configuration.
619 :return:
620 The list of paths, where each path is a tuple of (option, value).
621 """
623 def _all_items(section: str) -> List[Tuple[str, str]]:
624 """Return all (key, value) pairs for a section, including duplicate keys."""
625 return [
626 (key, value)
627 for key, values in self._sections[section].items_all()
628 if key != "__name__"
629 for value in values
630 ]
632 paths = []
634 for section in self.sections():
635 if section == "include":
636 paths += _all_items(section)
638 match = CONDITIONAL_INCLUDE_REGEXP.search(section)
639 if match is None or self._repo is None:
640 continue
642 keyword = match.group(1)
643 value = match.group(2).strip()
645 if keyword in ["gitdir", "gitdir/i"]:
646 value = osp.expanduser(value)
647 git_dir = os.fspath(self._repo.git_dir) if self._repo.git_dir else None
648 if sys.platform == "win32":
649 git_dir = git_dir.replace("\\", "/") if git_dir else None
651 drive, _tail = osp.splitdrive(value)
652 if not drive and not any(value.startswith(s) for s in ["./", "/"]):
653 value = "**/" + value
654 if value.endswith("/"):
655 value += "**"
657 # Ensure that glob is always case insensitive if required.
658 if keyword.endswith("/i"):
659 value = re.sub(
660 r"[a-zA-Z]",
661 lambda m: f"[{m.group().lower()!r}{m.group().upper()!r}]",
662 value,
663 )
664 if git_dir and fnmatch.fnmatchcase(git_dir, value):
665 paths += _all_items(section)
667 elif keyword == "onbranch":
668 try:
669 branch_name = self._repo.active_branch.name
670 except TypeError:
671 # Ignore section if active branch cannot be retrieved.
672 continue
674 if fnmatch.fnmatchcase(branch_name, value):
675 paths += _all_items(section)
676 elif keyword == "hasconfig:remote.*.url":
677 for remote in self._repo.remotes:
678 if fnmatch.fnmatchcase(remote.url, value):
679 paths += _all_items(section)
680 break
681 return paths
683 def read(self) -> None: # type: ignore[override]
684 """Read the data stored in the files we have been initialized with.
686 This will ignore files that cannot be read, possibly leaving an empty
687 configuration.
689 :raise IOError:
690 If a file cannot be handled.
691 """
692 if self._is_initialized:
693 return
694 self._is_initialized = True
696 files_to_read: List[Union[PathLike, IO]] = [""]
697 if isinstance(self._file_or_files, (str, os.PathLike)):
698 # For str or Path, as str is a type of Sequence.
699 files_to_read = [self._file_or_files]
700 elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
701 # Could merge with above isinstance once runtime type known.
702 files_to_read = [self._file_or_files]
703 else: # For lists or tuples.
704 files_to_read = list(self._file_or_files)
705 # END ensure we have a copy of the paths to handle
707 files_to_read = [osp.abspath(path) if isinstance(path, (str, os.PathLike)) else path for path in files_to_read]
709 seen = set(files_to_read)
710 num_read_include_files = 0
711 while files_to_read:
712 file_path = files_to_read.pop(0)
713 file_ok = False
715 if hasattr(file_path, "seek"):
716 # Must be a file-object.
717 # TODO: Replace cast with assert to narrow type, once sure.
718 file_path = cast(IO[bytes], file_path)
719 self._read(file_path, file_path.name)
720 else:
721 try:
722 with open(file_path, "rb") as fp:
723 file_ok = True
724 self._read(fp, fp.name)
725 except IOError:
726 continue
728 # Read includes and append those that we didn't handle yet. We expect all
729 # paths to be normalized and absolute (and will ensure that is the case).
730 if self._has_includes():
731 for _, include_path in self._included_paths():
732 if include_path.startswith("~"):
733 include_path = osp.expanduser(include_path)
734 if not osp.isabs(include_path):
735 if not file_ok:
736 continue
737 # END ignore relative paths if we don't know the configuration file path
738 file_path = cast(PathLike, file_path)
739 assert osp.isabs(file_path), "Need absolute paths to be sure our cycle checks will work"
740 include_path = osp.join(osp.dirname(file_path), include_path)
741 # END make include path absolute
742 include_path = osp.normpath(include_path)
743 if include_path in seen or not os.access(include_path, os.R_OK):
744 continue
745 seen.add(include_path)
746 # Insert included file to the top to be considered first.
747 files_to_read.insert(0, include_path)
748 num_read_include_files += 1
749 # END each include path in configuration file
750 # END handle includes
751 # END for each file object to read
753 # If there was no file included, we can safely write back (potentially) the
754 # configuration file without altering its meaning.
755 if num_read_include_files == 0:
756 self._merge_includes = False
758 def _write(self, fp: IO) -> None:
759 """Write an .ini-format representation of the configuration state in
760 git compatible format."""
762 def write_section(name: str, section_dict: _OMD) -> None:
763 fp.write(("[%s]\n" % name).encode(defenc))
765 values: Sequence[str] # Runtime only gets str in tests, but should be whatever _OMD stores.
766 v: str
767 for key, values in section_dict.items_all():
768 if key == "__name__":
769 continue
771 for v in values:
772 value = self._value_to_string(v)
773 if any(char in value for char in '\n\t\b\\"#;') or value[:1].isspace() or value[-1:].isspace():
774 value = value.replace("\\", "\\\\").replace('"', '\\"')
775 value = '"%s\\\n"' % value.replace("\n", "\\n").replace("\t", "\\t").replace("\b", "\\b")
776 fp.write(("\t%s = %s\n" % (key, value)).encode(defenc))
777 # END if key is not __name__
779 # END section writing
781 if self._defaults:
782 write_section(cp.DEFAULTSECT, self._defaults)
783 value: _OMD
785 for name, value in self._sections.items():
786 write_section(name, value)
788 def items(self, section_name: str) -> List[Tuple[str, str]]: # type: ignore[override]
789 """:return: list((option, value), ...) pairs of all items in the given section"""
790 return [(k, v) for k, v in super().items(section_name) if k != "__name__"]
792 def items_all(self, section_name: str) -> List[Tuple[str, List[str]]]:
793 """:return: list((option, [values...]), ...) pairs of all items in the given section"""
794 rv = _OMD(self._defaults)
796 for k, vs in self._sections[section_name].items_all():
797 if k == "__name__":
798 continue
800 if k in rv and rv.getall(k) == vs:
801 continue
803 for v in vs:
804 rv.add(k, v)
806 return rv.items_all()
808 @needs_values
809 def write(self) -> None:
810 """Write changes to our file, if there are changes at all.
812 :raise IOError:
813 If this is a read-only writer instance or if we could not obtain a file
814 lock.
815 """
816 self._assure_writable("write")
817 if not self._dirty:
818 return
820 if isinstance(self._file_or_files, (list, tuple)):
821 raise AssertionError(
822 "Cannot write back if there is not exactly a single file to write to, have %i files"
823 % len(self._file_or_files)
824 )
825 # END assert multiple files
827 if self._has_includes():
828 _logger.debug(
829 "Skipping write-back of configuration file as include files were merged in."
830 + "Set merge_includes=False to prevent this."
831 )
832 return
833 # END stop if we have include files
835 sections: List[_OMD] = [self._defaults]
836 section: _OMD
837 stored_section: _OMD
838 values: List[Any]
839 raw_value: Any
840 for _, stored_section in self._sections.items():
841 sections.append(stored_section)
842 for section in sections:
843 for key, values in section.items_all():
844 if key != "__name__":
845 for raw_value in values:
846 if "\r" in self._value_to_string(raw_value) or "\x00" in self._value_to_string(raw_value):
847 raise ValueError("Git config values must not contain CR or NUL")
849 fp = self._file_or_files
851 # We have a physical file on disk, so get a lock.
852 is_file_lock = isinstance(fp, (str, os.PathLike, IOBase)) # TODO: Use PathLike (having dropped 3.5).
853 if is_file_lock and self._lock is not None: # Else raise error?
854 self._lock._obtain_lock()
856 if not hasattr(fp, "seek"):
857 fp = cast(PathLike, fp)
858 with open(fp, "wb") as fp_open:
859 self._write(fp_open)
860 else:
861 fp = cast("BytesIO", fp)
862 fp.seek(0)
863 # Make sure we do not overwrite into an existing file.
864 if hasattr(fp, "truncate"):
865 fp.truncate()
866 self._write(fp)
868 def _assure_writable(self, method_name: str) -> None:
869 if self.read_only:
870 raise IOError("Cannot execute non-constant method %s.%s" % (self, method_name))
872 def add_section(self, section: "cp._SectionName") -> None:
873 """Assures added options will stay in order."""
874 self._assure_config_name_safe(section, "section")
875 return super().add_section(section)
877 @property
878 def read_only(self) -> bool:
879 """:return: ``True`` if this instance may change the configuration file"""
880 return self._read_only
882 # FIXME: Figure out if default or return type can really include bool.
883 def get_value(
884 self,
885 section: str,
886 option: str,
887 default: Union[int, float, str, bool, None] = None,
888 ) -> Union[int, float, str, bool]:
889 """Get an option's value.
891 If multiple values are specified for this option in the section, the last one
892 specified is returned.
894 :param default:
895 If not ``None``, the given default value will be returned in case the option
896 did not exist.
898 :return:
899 A properly typed value, either int, float or string
901 :raise TypeError:
902 In case the value could not be understood.
903 Otherwise the exceptions known to the ConfigParser will be raised.
904 """
905 try:
906 valuestr = self.get(section, option)
907 except Exception:
908 if default is not None:
909 return default
910 raise
912 return self._string_to_value(valuestr)
914 def get_values(
915 self,
916 section: str,
917 option: str,
918 default: Union[int, float, str, bool, None] = None,
919 ) -> List[Union[int, float, str, bool]]:
920 """Get an option's values.
922 If multiple values are specified for this option in the section, all are
923 returned.
925 :param default:
926 If not ``None``, a list containing the given default value will be returned
927 in case the option did not exist.
929 :return:
930 A list of properly typed values, either int, float or string
932 :raise TypeError:
933 In case the value could not be understood.
934 Otherwise the exceptions known to the ConfigParser will be raised.
935 """
936 try:
937 self.sections()
938 lst = self._sections[section].getall(option)
939 except Exception:
940 if default is not None:
941 return [default]
942 raise
944 return [self._string_to_value(valuestr) for valuestr in lst]
946 def _string_to_value(self, valuestr: str) -> Union[int, float, str, bool]:
947 types = (int, float)
948 for numtype in types:
949 try:
950 val = numtype(valuestr)
951 # truncated value ?
952 if val != float(valuestr):
953 continue
954 return val
955 except (ValueError, TypeError):
956 continue
957 # END for each numeric type
959 # Try boolean values as git uses them.
960 vl = valuestr.lower()
961 if vl == "false":
962 return False
963 if vl == "true":
964 return True
966 if not isinstance(valuestr, str):
967 raise TypeError(
968 "Invalid value type: only int, long, float and str are allowed",
969 valuestr,
970 )
972 return valuestr
974 def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str:
975 if isinstance(value, (int, float, bool)):
976 return str(value)
977 return force_text(value)
979 def _value_to_string_safe(self, value: Union[str, bytes, int, float, bool]) -> str:
980 value_str = self._value_to_string(value)
981 if UNSAFE_CONFIG_CHARS_RE.search(value_str):
982 raise ValueError("Git config values must not contain CR, LF, or NUL")
983 return value_str
985 def _assure_config_name_safe(self, name: "cp._SectionName", label: str) -> None:
986 if isinstance(name, str) and UNSAFE_CONFIG_CHARS_RE.search(name):
987 raise ValueError("Git config %s names must not contain CR, LF, or NUL" % label)
988 if label == "option" and isinstance(name, str) and not VALID_CONFIG_OPTION_NAME_RE.fullmatch(name):
989 raise ValueError("Git config option names may contain only letters, digits, '-', '_', or '.'")
990 if label == "section" and isinstance(name, str):
991 in_quotes = False
992 escaped = False
993 for index, char in enumerate(name):
994 if escaped:
995 escaped = False
996 elif in_quotes and char == "\\":
997 escaped = True
998 elif char == '"':
999 if not in_quotes and (index == 0 or name[index - 1] not in " \t"):
1000 raise ValueError("Git config quoted subsection names must begin after whitespace")
1001 in_quotes = not in_quotes
1002 elif char == "]" and not in_quotes:
1003 raise ValueError("Git config section names must not contain an unquoted closing bracket")
1004 if in_quotes:
1005 raise ValueError("Git config section names must not contain an unterminated quote")
1007 @needs_values
1008 @set_dirty_and_flush_changes
1009 def set(
1010 self,
1011 section: str,
1012 option: str,
1013 value: Union[str, bytes, int, float, bool, None] = None,
1014 ) -> None:
1015 self._assure_config_name_safe(section, "section")
1016 self._assure_config_name_safe(option, "option")
1017 if value is not None:
1018 value = self._value_to_string_safe(value)
1019 return super().set(section, option, value)
1021 @needs_values
1022 @set_dirty_and_flush_changes
1023 def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
1024 """Set the given option in section to the given value.
1026 This will create the section if required, and will not throw as opposed to the
1027 default ConfigParser ``set`` method.
1029 :param section:
1030 Name of the section in which the option resides or should reside.
1032 :param option:
1033 Name of the options whose value to set.
1035 :param value:
1036 Value to set the option to. It must be a string or convertible to a string.
1038 :return:
1039 This instance
1040 """
1041 self._assure_config_name_safe(section, "section")
1042 self._assure_config_name_safe(option, "option")
1043 value_str = self._value_to_string_safe(value)
1044 if not self.has_section(section):
1045 self.add_section(section)
1046 super().set(section, option, value_str)
1047 return self
1049 @needs_values
1050 @set_dirty_and_flush_changes
1051 def add_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
1052 """Add a value for the given option in section.
1054 This will create the section if required, and will not throw as opposed to the
1055 default ConfigParser ``set`` method. The value becomes the new value of the
1056 option as returned by :meth:`get_value`, and appends to the list of values
1057 returned by :meth:`get_values`.
1059 :param section:
1060 Name of the section in which the option resides or should reside.
1062 :param option:
1063 Name of the option.
1065 :param value:
1066 Value to add to option. It must be a string or convertible to a string.
1068 :return:
1069 This instance
1070 """
1071 self._assure_config_name_safe(section, "section")
1072 self._assure_config_name_safe(option, "option")
1073 value_str = self._value_to_string_safe(value)
1074 if not self.has_section(section):
1075 self.add_section(section)
1076 self._sections[section].add(option, value_str)
1077 return self
1079 def rename_section(self, section: str, new_name: str) -> "GitConfigParser":
1080 """Rename the given section to `new_name`.
1082 :raise ValueError:
1083 If:
1085 * `section` doesn't exist.
1086 * A section with `new_name` does already exist.
1088 :return:
1089 This instance
1090 """
1091 if not self.has_section(section):
1092 raise ValueError("Source section '%s' doesn't exist" % section)
1093 self._assure_config_name_safe(new_name, "section")
1094 if self.has_section(new_name):
1095 raise ValueError("Destination section '%s' already exists" % new_name)
1097 super().add_section(new_name)
1098 new_section = self._sections[new_name]
1099 for k, vs in self.items_all(section):
1100 new_section.setall(k, vs)
1101 # END for each value to copy
1103 # This call writes back the changes, which is why we don't have the respective
1104 # decorator.
1105 self.remove_section(section)
1106 return self