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
13import inspect
14import logging
15import os
16import os.path as osp
17import re
18import sys
19from functools import wraps
20from io import BufferedReader, IOBase
22# typing-------------------------------------------------------
23from typing import (
24 IO,
25 TYPE_CHECKING,
26 Any,
27 Callable,
28 Dict,
29 Generic,
30 List,
31 OrderedDict,
32 Sequence,
33 Tuple,
34 TypeVar,
35 Union,
36 cast,
37)
39from git.compat import defenc, force_text
40from git.types import _T, ConfigLevels_Tup, Lit_config_levels, PathLike, assert_never
41from git.util import LockFile
43if TYPE_CHECKING:
44 from io import BytesIO
46 from git.repo.base import Repo
48T_ConfigParser = TypeVar("T_ConfigParser", bound="GitConfigParser")
49T_OMD_value = TypeVar("T_OMD_value", str, bytes, int, float, bool, None)
51OrderedDict_OMD = OrderedDict[str, List[T_OMD_value]]
53# -------------------------------------------------------------
55_logger = logging.getLogger(__name__)
57CONFIG_LEVELS: ConfigLevels_Tup = ("system", "user", "global", "repository")
58"""The configuration level of a configuration file."""
60CONDITIONAL_INCLUDE_REGEXP = re.compile(r"(?<=includeif )\"(gitdir|gitdir/i|onbranch|hasconfig:remote\.\*\.url):(.+)\"")
61"""Section pattern to detect conditional includes.
63See: https://git-scm.com/docs/git-config#_conditional_includes
64"""
66UNSAFE_CONFIG_CHARS_RE = re.compile(r"[\r\n\x00]")
67"""Characters that cannot be safely written in config names or values."""
69VALID_CONFIG_OPTION_NAME_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
70"""Pattern for option names that can be written without changing config syntax."""
73class MetaParserBuilder(abc.ABCMeta): # noqa: B024
74 """Utility class wrapping base-class methods into decorators that assure read-only
75 properties."""
77 def __new__(cls, name: str, bases: Tuple, clsdict: Dict[str, Any]) -> "MetaParserBuilder":
78 """Equip all base-class methods with a needs_values decorator, and all non-const
79 methods with a :func:`set_dirty_and_flush_changes` decorator in addition to
80 that.
81 """
82 kmm = "_mutating_methods_"
83 if kmm in clsdict:
84 mutating_methods = clsdict[kmm]
85 for base in bases:
86 methods = (t for t in inspect.getmembers(base, inspect.isroutine) if not t[0].startswith("_"))
87 for method_name, method in methods:
88 if method_name in clsdict:
89 continue
90 method_with_values = needs_values(method)
91 if method_name in mutating_methods:
92 method_with_values = set_dirty_and_flush_changes(method_with_values)
93 # END mutating methods handling
95 clsdict[method_name] = method_with_values
96 # END for each name/method pair
97 # END for each base
98 # END if mutating methods configuration is set
100 new_type = super().__new__(cls, name, bases, clsdict)
101 return new_type
104def needs_values(func: Callable[..., _T]) -> Callable[..., _T]:
105 """Return a method for ensuring we read values (on demand) before we try to access
106 them."""
108 @wraps(func)
109 def assure_data_present(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
110 self.read()
111 return func(self, *args, **kwargs)
113 # END wrapper method
114 return assure_data_present
117def set_dirty_and_flush_changes(non_const_func: Callable[..., _T]) -> Callable[..., _T]:
118 """Return a method that checks whether given non constant function may be called.
120 If so, the instance will be set dirty. Additionally, we flush the changes right to
121 disk.
122 """
124 def flush_changes(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
125 rval = non_const_func(self, *args, **kwargs)
126 self._dirty = True
127 self.write()
128 return rval
130 # END wrapper method
131 flush_changes.__name__ = non_const_func.__name__
132 return flush_changes
135class SectionConstraint(Generic[T_ConfigParser]):
136 """Constrains a ConfigParser to only option commands which are constrained to
137 always use the section we have been initialized with.
139 It supports all ConfigParser methods that operate on an option.
141 :note:
142 If used as a context manager, will release the wrapped ConfigParser.
143 """
145 __slots__ = ("_config", "_section_name")
147 _valid_attrs_ = (
148 "get_value",
149 "set_value",
150 "get",
151 "set",
152 "getint",
153 "getfloat",
154 "getboolean",
155 "has_option",
156 "remove_section",
157 "remove_option",
158 "options",
159 )
161 def __init__(self, config: T_ConfigParser, section: str) -> None:
162 self._config = config
163 self._section_name = section
165 def __del__(self) -> None:
166 # Yes, for some reason, we have to call it explicitly for it to work in PY3 !
167 # Apparently __del__ doesn't get call anymore if refcount becomes 0
168 # Ridiculous ... .
169 self._config.release()
171 def __getattr__(self, attr: str) -> Any:
172 if attr in self._valid_attrs_:
173 return lambda *args, **kwargs: self._call_config(attr, *args, **kwargs)
174 return super().__getattribute__(attr)
176 def _call_config(self, method: str, *args: Any, **kwargs: Any) -> Any:
177 """Call the configuration at the given method which must take a section name as
178 first argument."""
179 return getattr(self._config, method)(self._section_name, *args, **kwargs)
181 @property
182 def config(self) -> T_ConfigParser:
183 """return: ConfigParser instance we constrain"""
184 return self._config
186 def release(self) -> None:
187 """Equivalent to :meth:`GitConfigParser.release`, which is called on our
188 underlying parser instance."""
189 return self._config.release()
191 def __enter__(self) -> "SectionConstraint[T_ConfigParser]":
192 self._config.__enter__()
193 return self
195 def __exit__(self, exception_type: str, exception_value: str, traceback: str) -> None:
196 self._config.__exit__(exception_type, exception_value, traceback)
199def _normalize_name(name: str) -> str:
200 """Fold section and option names, leaving quoted subsections unchanged."""
201 prefix, separator, subsection = name.partition('"')
202 return prefix.lower() + separator + subsection
205class _OMD(OrderedDict_OMD):
206 """Ordered multi-dict matching config names while retaining their first spelling."""
208 def __init__(self, *args: Any, **kwargs: Any) -> None:
209 self._keymap: Dict[str, str] = {}
210 super().__init__(*args, **kwargs)
212 def _key(self, key: str) -> str:
213 stored = self._keymap.get(_normalize_name(key), key)
214 return stored if super().__contains__(stored) else key
216 def __contains__(self, key: object) -> bool:
217 return isinstance(key, str) and super().__contains__(self._key(key))
219 def __delitem__(self, key: str) -> None:
220 super().__delitem__(self._key(key))
221 del self._keymap[_normalize_name(key)]
223 def __setitem__(self, key: str, value: _T) -> None:
224 self.setall(key, [value])
226 def clear(self) -> None:
227 super().clear()
228 self._keymap.clear()
230 def add(self, key: str, value: Any) -> None:
231 if key not in self:
232 self[key] = value
233 return
235 self.getall(key).append(value)
237 def setall(self, key: str, values: List[_T]) -> None:
238 key = self._key(key)
239 super().__setitem__(key, values)
240 self._keymap[_normalize_name(key)] = key
242 def __getitem__(self, key: str) -> Any:
243 return super().__getitem__(self._key(key))[-1]
245 def getlast(self, key: str) -> Any:
246 return self[key]
248 def setlast(self, key: str, value: Any) -> None:
249 if key not in self:
250 self[key] = value
251 return
253 self.getall(key)[-1] = value
255 def get(self, key: str, default: Union[_T, None] = None) -> Union[_T, None]: # type: ignore[override]
256 return super().get(self._key(key), [default])[-1]
258 def getall(self, key: str) -> List[_T]:
259 return super().__getitem__(self._key(key))
261 def items(self) -> List[Tuple[str, _T]]: # type: ignore[override]
262 """List of (key, last value for key)."""
263 return [(k, self[k]) for k in self]
265 def items_all(self) -> List[Tuple[str, List[_T]]]:
266 """List of (key, list of values for key)."""
267 return [(k, self.getall(k)) for k in self]
270def get_config_path(config_level: Lit_config_levels) -> str:
271 # We do not support an absolute path of the gitconfig on Windows.
272 # Use the global config instead.
273 if sys.platform == "win32" and config_level == "system":
274 config_level = "global"
276 if config_level == "system":
277 return "/etc/gitconfig"
278 elif config_level == "user":
279 config_home = os.environ.get("XDG_CONFIG_HOME") or osp.join(os.environ.get("HOME", "~"), ".config")
280 return osp.normpath(osp.expanduser(osp.join(config_home, "git", "config")))
281 elif config_level == "global":
282 return osp.normpath(osp.expanduser("~/.gitconfig"))
283 elif config_level == "repository":
284 raise ValueError("No repo to get repository configuration from. Use Repo._get_config_path")
285 else:
286 # Should not reach here. Will raise ValueError if does. Static typing will warn
287 # about missing elifs.
288 assert_never( # type: ignore[unreachable]
289 config_level,
290 ValueError(f"Invalid configuration level: {config_level!r}"),
291 )
294class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder):
295 """Implements specifics required to read git style configuration files.
297 This variation behaves much like the :manpage:`git-config(1)` command, such that the
298 configuration will be read on demand based on the filepath given during
299 initialization.
301 The changes will automatically be written once the instance goes out of scope, but
302 can be triggered manually as well.
304 The configuration file will be locked if you intend to change values preventing
305 other instances to write concurrently.
307 :note:
308 Section and option names are case-insensitive; quoted subsection names are
309 case-sensitive. Names retain their first spelling when enumerated or written.
310 Case variants are merged, preserving all values in the order they are read.
312 :note:
313 If used as a context manager, this will release the locked file.
315 :note:
316 Options without a value are stored as ``None`` and written without ``=``.
317 :meth:`get_value` and :meth:`get_values` return an empty string for them,
318 while :meth:`getboolean` returns ``True``. An explicit empty value is
319 stored as an empty string and reads as ``False`` with :meth:`getboolean`.
320 """
322 # { Configuration
323 t_lock = LockFile
324 """The lock type determines the type of lock to use in new configuration readers.
326 They must be compatible to the :class:`~git.util.LockFile` interface.
327 A suitable alternative would be the :class:`~git.util.BlockingLockFile`.
328 """
330 re_comment = re.compile(r"^\s*[#;]")
331 # } END configuration
333 optvalueonly_source = r"\s*(?P<option>[^:=\s#;][^:=#;]*)"
335 OPTVALUEONLY = re.compile(optvalueonly_source)
337 # The option name class [^:=#;]* already consumes any spaces up to the ":" or "=",
338 # so a second \s* before the indicator would overlap it and backtrack quadratically
339 # on a line that never reaches an indicator (for example a key followed by a long
340 # whitespace run). Drop the redundant \s*; the name is right-stripped after parsing.
341 OPTCRE = re.compile(optvalueonly_source + r"(?P<vi>[:=])\s*" + r"(?P<value>.*)$")
343 del optvalueonly_source
345 _mutating_methods_ = ("add_section", "remove_section", "remove_option", "set")
346 """Names of :class:`~configparser.RawConfigParser` methods able to change the
347 instance."""
349 def __init__(
350 self,
351 file_or_files: Union[None, PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = None,
352 read_only: bool = True,
353 merge_includes: bool = True,
354 config_level: Union[Lit_config_levels, None] = None,
355 repo: Union["Repo", None] = None,
356 ) -> None:
357 """Initialize a configuration reader to read the given `file_or_files` and to
358 possibly allow changes to it by setting `read_only` False.
360 :param file_or_files:
361 A file path or file object, or a sequence of possibly more than one of them.
363 :param read_only:
364 If ``True``, the ConfigParser may only read the data, but not change it.
365 If ``False``, only a single file path or file object may be given. We will
366 write back the changes when they happen, or when the ConfigParser is
367 released. This will not happen if other configuration files have been
368 included.
370 :param merge_includes:
371 If ``True``, we will read files mentioned in ``[include]`` sections and
372 merge their contents into ours. This makes it impossible to write back an
373 individual configuration file. Thus, if you want to modify a single
374 configuration file, turn this off to leave the original dataset unaltered
375 when reading it.
377 :param repo:
378 Reference to repository to use if ``[includeIf]`` sections are found in
379 configuration files.
380 """
381 cp.RawConfigParser.__init__(self, dict_type=cast(Any, _OMD), allow_no_value=True)
382 self._dict: Callable[..., _OMD]
383 self._defaults: _OMD
384 self._sections: _OMD
386 # Used in Python 3. Needs to stay in sync with sections for underlying
387 # implementation to work.
388 if not hasattr(self, "_proxies"):
389 self._proxies = self._dict()
391 if file_or_files is not None:
392 self._file_or_files: Union[PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = file_or_files
393 else:
394 if config_level is None:
395 if read_only:
396 self._file_or_files = [
397 get_config_path(cast(Lit_config_levels, f)) for f in CONFIG_LEVELS if f != "repository"
398 ]
399 else:
400 raise ValueError("No configuration level or configuration files specified")
401 else:
402 self._file_or_files = [get_config_path(config_level)]
404 self._read_only = read_only
405 self._dirty = False
406 self._is_initialized = False
407 self._merge_includes = merge_includes
408 self._repo = repo
409 self._lock: Union["LockFile", None] = None
410 self._acquire_lock()
412 def _acquire_lock(self) -> None:
413 if not self._read_only:
414 if not self._lock:
415 if isinstance(self._file_or_files, (str, os.PathLike)):
416 file_or_files = self._file_or_files
417 elif isinstance(self._file_or_files, (tuple, list, Sequence)):
418 raise ValueError(
419 "Write-ConfigParsers can operate on a single file only, multiple files have been passed"
420 )
421 else:
422 file_or_files = self._file_or_files.name
424 # END get filename from handle/stream
425 # Initialize lock base - we want to write.
426 self._lock = self.t_lock(file_or_files)
427 # END lock check
429 self._lock._obtain_lock()
430 # END read-only check
432 def __del__(self) -> None:
433 """Write pending changes if required and release locks."""
434 # NOTE: Only consistent in Python 2.
435 self.release()
437 def __enter__(self) -> "GitConfigParser":
438 self._acquire_lock()
439 return self
441 def __exit__(self, *args: Any) -> None:
442 self.release()
444 def release(self) -> None:
445 """Flush changes and release the configuration write lock. This instance must
446 not be used anymore afterwards.
448 In Python 3, it's required to explicitly release locks and flush changes, as
449 ``__del__`` is not called deterministically anymore.
450 """
451 # Checking for the lock here makes sure we do not raise during write()
452 # in case an invalid parser was created who could not get a lock.
453 if self.read_only or (self._lock and not self._lock._has_lock()):
454 return
456 try:
457 self.write()
458 except OSError:
459 _logger.error("Exception during destruction of GitConfigParser", exc_info=True)
460 except ReferenceError:
461 # This happens in Python 3... and usually means that some state cannot be
462 # written as the sections dict cannot be iterated. This usually happens when
463 # the interpreter is shutting down. Can it be fixed?
464 pass
465 finally:
466 if self._lock is not None:
467 self._lock._release_lock()
469 def optionxform(self, optionstr: str) -> str:
470 """Do not transform options in any way when writing."""
471 return optionstr
473 def _read(self, fp: Union[BufferedReader, IO[bytes]], fpname: str) -> None:
474 """Originally a direct copy of the Python 2.4 version of
475 :meth:`RawConfigParser._read <configparser.RawConfigParser._read>`, to ensure it
476 uses ordered dicts.
478 The ordering bug was fixed in Python 2.4, and dict itself keeps ordering since
479 Python 3.7. This has some other changes, especially that it ignores initial
480 whitespace, since git uses tabs. (Big comments are removed to be more compact.)
481 """
482 cursect = None # None, or a dictionary.
483 optname = None
484 lineno = 0
485 is_multi_line = False
486 e = None # None, or an exception.
488 def string_decode(v: str) -> str:
489 if v and v.endswith("\\"):
490 v = v[:-1]
491 # END cut trailing escapes to prevent decode error
493 escapes = {"b": "\b", "n": "\n", "r": "\r", "t": "\t", '"': '"', "\\": "\\"}
494 return re.sub(r"\\(.)", lambda match: escapes.get(match.group(1), match.group(0)), v)
496 # END string_decode
498 def is_line_continuation(value: str) -> bool:
499 quoted = escaped = False
500 for char in value:
501 if escaped:
502 escaped = False
503 elif char == "\\":
504 escaped = True
505 elif char == '"':
506 quoted = not quoted
507 elif char in "#;" and not quoted:
508 return False
509 return escaped
511 def strip_inline_comment(value: str) -> Tuple[str, bool]:
512 """Cut an unquoted ``#`` or ``;`` comment and report whether a quote is open.
514 Quoting and backslash escapes are honoured, so a ``#`` inside a quoted
515 value is literal and an unterminated quote swallows the rest of the line.
516 """
517 quoted = escaped = False
518 for index, char in enumerate(value):
519 if escaped:
520 escaped = False
521 elif char == "\\":
522 escaped = True
523 elif char == '"':
524 quoted = not quoted
525 elif char in "#;" and not quoted:
526 return value[:index], False
527 return value, quoted
529 def parse_value(value: str) -> str:
530 parsed: List[str] = []
531 whitespace: List[str] = []
532 quoted = escaped = False
533 escapes = {"b": "\b", "n": "\n", "t": "\t", '"': '"', "\\": "\\"}
534 for char in value:
535 if escaped:
536 parsed.append(escapes.get(char, "\\" + char))
537 escaped = False
538 continue
539 if char.isspace() and not quoted:
540 if parsed:
541 whitespace.append(char)
542 continue
543 if char in "#;" and not quoted:
544 break
545 parsed.extend(whitespace)
546 whitespace.clear()
547 if char == "\\":
548 escaped = True
549 elif char == '"':
550 quoted = not quoted
551 else:
552 parsed.append(char)
553 return "".join(parsed)
555 while True:
556 # We assume to read binary!
557 raw_line = fp.readline()
558 if not raw_line:
559 break
560 if lineno == 0 and raw_line.startswith(b"\xef\xbb\xbf"):
561 # A UTF-8 BOM is not part of the content. git skips it, so a
562 # config file written by a Windows editor still parses.
563 raw_line = raw_line[3:]
564 line = raw_line.decode(defenc)
565 lineno = lineno + 1
566 # Comment or blank line?
567 if line.strip() == "" or self.re_comment.match(line):
568 continue
569 if line.split(None, 1)[0].lower() == "rem" and line[0] in "rR":
570 # No leading whitespace.
571 continue
573 # Is it a section header?
574 mo = self.SECTCRE.match(line.strip())
575 if not is_multi_line and mo:
576 sectname: str = mo.group("header").strip()
577 if sectname in self._sections:
578 cursect = self._sections[sectname]
579 elif sectname == cp.DEFAULTSECT:
580 cursect = self._defaults
581 else:
582 cursect = self._dict((("__name__", sectname),))
583 self._sections[sectname] = cursect
584 self._proxies[sectname] = None
585 # So sections can't start with a continuation line.
586 optname = None
587 # No section header in the file?
588 elif cursect is None:
589 raise cp.MissingSectionHeaderError(fpname, lineno, line)
590 # An option line?
591 elif not is_multi_line:
592 mo = self.OPTCRE.match(line)
593 if mo:
594 # We might just have handled the last line, which could contain a quotation we want to remove.
595 optname, vi, optval = mo.group("option", "vi", "value")
596 optname = self.optionxform(optname.rstrip())
598 optval, quote_open = strip_inline_comment(optval)
599 optval = optval.strip()
601 if len(optval) < 2 or optval[0] != '"':
602 # Does not open quoting.
603 # A value ending in an odd number of backslashes
604 # continues on the next line, exactly as git does: the
605 # final backslash and the newline are removed and the
606 # next line is appended before the complete value is
607 # parsed. An even number means the last backslash is
608 # escaped and the value ends there.
609 continued = False
610 while True:
611 if not is_line_continuation(optval):
612 break
613 continuation = fp.readline()
614 if not continuation:
615 # Backslash at end of file: git drops it.
616 optval = optval[:-1]
617 break
618 lineno = lineno + 1
619 joined = continuation.decode(defenc)
620 while joined.endswith("\n") or joined.endswith("\r"):
621 joined = joined[:-1]
622 optval = optval[:-1] + joined
623 continued = True
624 if continued:
625 optval = parse_value(optval)
626 elif quote_open:
627 # Opens quoting and does not close: appears to start multi-line quoting.
628 is_multi_line = True
629 optval = string_decode(optval[1:])
630 elif re.search(r'(?:^|[^\\])(?:\\\\)*"', optval[1:-1]):
631 # Preserve values containing additional unescaped quotes.
632 pass
633 else:
634 # Opens and closes quoting.
635 optval = string_decode(optval[1:-1])
637 # Preserves multiple values for duplicate optnames.
638 cursect.add(optname, optval)
639 else:
640 # A valueless option is an implicit boolean true, not an empty value.
641 mo = self.OPTVALUEONLY.fullmatch(line)
642 if mo:
643 optname = self.optionxform(mo.group("option").rstrip())
644 cursect.add(optname, None)
645 else:
646 if not e:
647 e = cp.ParsingError(fpname)
648 e.append(lineno, repr(line))
649 continue
650 else:
651 line = line.rstrip()
652 if line.endswith('"'):
653 is_multi_line = False
654 line = line[:-1]
655 # END handle quotations
656 optval = cursect.getlast(optname)
657 cursect.setlast(optname, optval + string_decode(line))
658 # END parse section or option
659 # END while reading
661 # If any parsing errors occurred, raise an exception.
662 if e:
663 raise e
665 def _has_includes(self) -> Union[bool, int]:
666 return self._merge_includes and len(self._included_paths())
668 def _included_paths(self) -> List[Tuple[str, str]]:
669 """List all paths that must be included to configuration.
671 :return:
672 The list of paths, where each path is a tuple of (option, value).
673 """
675 def _all_items(section: str) -> List[Tuple[str, str]]:
676 """Return all (key, value) pairs for a section, including duplicate keys."""
677 return [
678 (key, value)
679 for key, values in self._sections[section].items_all()
680 if key != "__name__"
681 for value in values
682 if value is not None
683 ]
685 paths = []
687 for section in self.sections():
688 normalized_section = _normalize_name(section)
689 if normalized_section == "include":
690 paths += _all_items(section)
692 match = CONDITIONAL_INCLUDE_REGEXP.search(normalized_section)
693 if match is None or self._repo is None:
694 continue
696 keyword = match.group(1)
697 value = match.group(2).strip()
699 if keyword in ["gitdir", "gitdir/i"]:
700 value = osp.expanduser(value)
701 git_dir = os.fspath(self._repo.git_dir) if self._repo.git_dir else None
702 if sys.platform == "win32":
703 git_dir = git_dir.replace("\\", "/") if git_dir else None
705 drive, _tail = osp.splitdrive(value)
706 if not drive and not any(value.startswith(s) for s in ["./", "/"]):
707 value = "**/" + value
708 if value.endswith("/"):
709 value += "**"
711 # Ensure that glob is always case insensitive if required.
712 if keyword.endswith("/i"):
713 value = re.sub(
714 r"[a-zA-Z]",
715 lambda m: f"[{m.group().lower()!r}{m.group().upper()!r}]",
716 value,
717 )
718 if git_dir and fnmatch.fnmatchcase(git_dir, value):
719 paths += _all_items(section)
721 elif keyword == "onbranch":
722 try:
723 branch_name = self._repo.active_branch.name
724 except TypeError:
725 # Ignore section if active branch cannot be retrieved.
726 continue
728 if fnmatch.fnmatchcase(branch_name, value):
729 paths += _all_items(section)
730 elif keyword == "hasconfig:remote.*.url":
731 for remote in self._repo.remotes:
732 if fnmatch.fnmatchcase(remote.url, value):
733 paths += _all_items(section)
734 break
735 return paths
737 def read(self) -> None: # type: ignore[override]
738 """Read the data stored in the files we have been initialized with.
740 This will ignore files that cannot be read, possibly leaving an empty
741 configuration.
743 :raise IOError:
744 If a file cannot be handled.
745 """
746 if self._is_initialized:
747 return
748 self._is_initialized = True
750 files_to_read: List[Union[PathLike, IO]] = [""]
751 if isinstance(self._file_or_files, (str, os.PathLike)):
752 # For str or Path, as str is a type of Sequence.
753 files_to_read = [self._file_or_files]
754 elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
755 # Could merge with above isinstance once runtime type known.
756 files_to_read = [self._file_or_files]
757 else: # For lists or tuples.
758 files_to_read = list(self._file_or_files)
759 # END ensure we have a copy of the paths to handle
761 files_to_read = [osp.abspath(path) if isinstance(path, (str, os.PathLike)) else path for path in files_to_read]
763 seen = set(files_to_read)
764 num_read_include_files = 0
765 while files_to_read:
766 file_path = files_to_read.pop(0)
767 file_ok = False
769 if hasattr(file_path, "seek"):
770 # Must be a file-object.
771 # TODO: Replace cast with assert to narrow type, once sure.
772 file_path = cast(IO[bytes], file_path)
773 self._read(file_path, file_path.name)
774 else:
775 try:
776 with open(file_path, "rb") as fp:
777 file_ok = True
778 self._read(fp, fp.name)
779 except OSError:
780 continue
782 # Read includes and append those that we didn't handle yet. We expect all
783 # paths to be normalized and absolute (and will ensure that is the case).
784 if self._has_includes():
785 for _, include_path in self._included_paths():
786 if include_path.startswith("~"):
787 include_path = osp.expanduser(include_path)
788 if not osp.isabs(include_path):
789 if not file_ok:
790 continue
791 # END ignore relative paths if we don't know the configuration file path
792 file_path = cast(PathLike, file_path)
793 assert osp.isabs(file_path), "Need absolute paths to be sure our cycle checks will work"
794 include_path = osp.join(osp.dirname(file_path), include_path)
795 # END make include path absolute
796 include_path = osp.normpath(include_path)
797 if include_path in seen or not os.access(include_path, os.R_OK):
798 continue
799 seen.add(include_path)
800 # Insert included file to the top to be considered first.
801 files_to_read.insert(0, include_path)
802 num_read_include_files += 1
803 # END each include path in configuration file
804 # END handle includes
805 # END for each file object to read
807 # If there was no file included, we can safely write back (potentially) the
808 # configuration file without altering its meaning.
809 if num_read_include_files == 0:
810 self._merge_includes = False
812 def _write(self, fp: IO) -> None:
813 """Write an .ini-format representation of the configuration state in
814 git compatible format."""
816 def write_section(name: str, section_dict: _OMD) -> None:
817 fp.write(("[%s]\n" % name).encode(defenc))
819 values: List[Any]
820 v: Any
821 for key, values in section_dict.items_all():
822 if key == "__name__":
823 continue
825 for v in values:
826 if v is None:
827 fp.write(("\t%s\n" % key).encode(defenc))
828 continue
829 value = self._value_to_string(v)
830 if any(char in value for char in '\n\t\b\\"#;') or value[:1].isspace() or value[-1:].isspace():
831 value = value.replace("\\", "\\\\").replace('"', '\\"')
832 value = '"%s\\\n"' % value.replace("\n", "\\n").replace("\t", "\\t").replace("\b", "\\b")
833 fp.write(("\t%s = %s\n" % (key, value)).encode(defenc))
834 # END if key is not __name__
836 # END section writing
838 if self._defaults:
839 write_section(cp.DEFAULTSECT, self._defaults)
840 value: _OMD
842 for name, value in self._sections.items():
843 write_section(name, value)
845 def items(self, section_name: str) -> List[Tuple[str, Union[str, None]]]: # type: ignore[override]
846 """:return: list((option, value), ...) pairs of all items in the given section"""
847 return [(k, v) for k, v in super().items(section_name) if k != "__name__"]
849 def items_all(self, section_name: str) -> List[Tuple[str, List[Union[str, None]]]]:
850 """:return: list((option, [values...]), ...) pairs of all items in the given section"""
851 rv = _OMD(self._defaults)
853 for k, vs in self._sections[section_name].items_all():
854 if k == "__name__":
855 continue
857 if k in rv and rv.getall(k) == vs:
858 continue
860 for v in vs:
861 rv.add(k, v)
863 return rv.items_all()
865 @needs_values
866 def write(self) -> None:
867 """Write changes to our file, if there are changes at all.
869 :raise IOError:
870 If this is a read-only writer instance or if we could not obtain a file
871 lock.
872 """
873 self._assure_writable("write")
874 if not self._dirty:
875 return
877 if isinstance(self._file_or_files, (list, tuple)):
878 raise AssertionError(
879 "Cannot write back if there is not exactly a single file to write to, have %i files"
880 % len(self._file_or_files)
881 )
882 # END assert multiple files
884 if self._has_includes():
885 _logger.debug(
886 "Skipping write-back of configuration file as include files were merged in."
887 + "Set merge_includes=False to prevent this."
888 )
889 return
890 # END stop if we have include files
892 sections: List[_OMD] = [self._defaults]
893 section: _OMD
894 stored_section: _OMD
895 values: List[Any]
896 raw_value: Any
897 for _, stored_section in self._sections.items():
898 sections.append(stored_section)
899 for section in sections:
900 for key, values in section.items_all():
901 if key != "__name__":
902 for raw_value in values:
903 if raw_value is None:
904 continue
905 if "\r" in self._value_to_string(raw_value) or "\x00" in self._value_to_string(raw_value):
906 raise ValueError("Git config values must not contain CR or NUL")
908 fp = self._file_or_files
910 # We have a physical file on disk, so get a lock.
911 is_file_lock = isinstance(fp, (str, os.PathLike, IOBase)) # TODO: Use PathLike (having dropped 3.5).
912 if is_file_lock and self._lock is not None: # Else raise error?
913 self._lock._obtain_lock()
915 if not hasattr(fp, "seek"):
916 fp = cast(PathLike, fp)
917 with open(fp, "wb") as fp_open:
918 self._write(fp_open)
919 else:
920 fp = cast("BytesIO", fp)
921 fp.seek(0)
922 # Make sure we do not overwrite into an existing file.
923 if hasattr(fp, "truncate"):
924 fp.truncate()
925 self._write(fp)
927 def _assure_writable(self, method_name: str) -> None:
928 if self.read_only:
929 raise OSError(f"Cannot execute non-constant method {self}.{method_name}")
931 def add_section(self, section: "cp._SectionName") -> None:
932 """Assures added options will stay in order."""
933 self._assure_config_name_safe(section, "section")
934 return super().add_section(section)
936 @property
937 def read_only(self) -> bool:
938 """:return: ``True`` if this instance may change the configuration file"""
939 return self._read_only
941 def get_value(
942 self,
943 section: str,
944 option: str,
945 default: Union[int, float, str, bool, None] = None,
946 ) -> Union[int, float, str, bool]:
947 """Get an option's value.
949 If multiple values are specified for this option in the section, the last one
950 specified is returned.
952 :param default:
953 If not ``None``, the given default value will be returned in case the option
954 did not exist.
956 :return:
957 A properly typed value, either int, float, string or bool
959 :raise TypeError:
960 In case the value could not be understood.
961 Otherwise the exceptions known to the ConfigParser will be raised.
962 """
963 try:
964 valuestr = self.get(section, option)
965 except Exception:
966 if default is not None:
967 return default
968 raise
970 return self._string_to_value(valuestr)
972 def get_values(
973 self,
974 section: str,
975 option: str,
976 default: Union[int, float, str, bool, None] = None,
977 ) -> List[Union[int, float, str, bool]]:
978 """Get an option's values.
980 If multiple values are specified for this option in the section, all are
981 returned.
983 :param default:
984 If not ``None``, a list containing the given default value will be returned
985 in case the option did not exist.
987 :return:
988 A list of properly typed values, either int, float, string or bool
990 :raise TypeError:
991 In case the value could not be understood.
992 Otherwise the exceptions known to the ConfigParser will be raised.
993 """
994 try:
995 self.sections()
996 lst = self._sections[section].getall(option)
997 except Exception:
998 if default is not None:
999 return [default]
1000 raise
1002 return [self._string_to_value(valuestr) for valuestr in lst]
1004 def _convert_to_boolean(self, value: Union[str, None]) -> bool:
1005 if value is None:
1006 return True
1007 if value == "":
1008 return False
1009 try:
1010 return self.BOOLEAN_STATES[value.lower()]
1011 except KeyError:
1012 raise ValueError("Not a boolean: %s" % value) from None
1014 def _string_to_value(self, valuestr: Union[str, None]) -> Union[int, float, str, bool]:
1015 if valuestr is None:
1016 return ""
1017 types = (int, float)
1018 for numtype in types:
1019 try:
1020 val = numtype(valuestr)
1021 # truncated value ?
1022 if val != float(valuestr):
1023 continue
1024 return val
1025 except (ValueError, TypeError):
1026 continue
1027 # END for each numeric type
1029 # Try boolean values as git uses them.
1030 vl = valuestr.lower()
1031 if vl in ("false", "no", "off"):
1032 return False
1033 if vl in ("true", "yes", "on"):
1034 return True
1036 if not isinstance(valuestr, str):
1037 raise TypeError(
1038 "Invalid value type: only int, long, float and str are allowed",
1039 valuestr,
1040 )
1042 return valuestr
1044 def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str:
1045 if isinstance(value, (int, float, bool)):
1046 return str(value)
1047 return force_text(value)
1049 def _value_to_string_safe(self, value: Union[str, bytes, int, float, bool]) -> str:
1050 value_str = self._value_to_string(value)
1051 if UNSAFE_CONFIG_CHARS_RE.search(value_str):
1052 raise ValueError("Git config values must not contain CR, LF, or NUL")
1053 return value_str
1055 def _assure_config_name_safe(self, name: "cp._SectionName", label: str) -> None:
1056 if isinstance(name, str) and UNSAFE_CONFIG_CHARS_RE.search(name):
1057 raise ValueError("Git config %s names must not contain CR, LF, or NUL" % label)
1058 if label == "option" and isinstance(name, str) and not VALID_CONFIG_OPTION_NAME_RE.fullmatch(name):
1059 raise ValueError("Git config option names may contain only letters, digits, '-', '_', or '.'")
1060 if label == "section" and isinstance(name, str):
1061 in_quotes = False
1062 escaped = False
1063 for index, char in enumerate(name):
1064 if escaped:
1065 escaped = False
1066 elif in_quotes and char == "\\":
1067 escaped = True
1068 elif char == '"':
1069 if not in_quotes and (index == 0 or name[index - 1] not in " \t"):
1070 raise ValueError("Git config quoted subsection names must begin after whitespace")
1071 in_quotes = not in_quotes
1072 elif char == "]" and not in_quotes:
1073 raise ValueError("Git config section names must not contain an unquoted closing bracket")
1074 if in_quotes:
1075 raise ValueError("Git config section names must not contain an unterminated quote")
1077 @needs_values
1078 @set_dirty_and_flush_changes
1079 def set(
1080 self,
1081 section: str,
1082 option: str,
1083 value: Union[str, bytes, int, float, bool, None] = None,
1084 ) -> None:
1085 self._assure_config_name_safe(section, "section")
1086 self._assure_config_name_safe(option, "option")
1087 if value is not None:
1088 value = self._value_to_string_safe(value)
1089 return super().set(section, option, value)
1091 @needs_values
1092 @set_dirty_and_flush_changes
1093 def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
1094 """Set the given option in section to the given value.
1096 This will create the section if required, and will not throw as opposed to the
1097 default ConfigParser ``set`` method.
1099 :param section:
1100 Name of the section in which the option resides or should reside.
1102 :param option:
1103 Name of the options whose value to set.
1105 :param value:
1106 Value to set the option to. It must be a string or convertible to a string.
1108 :return:
1109 This instance
1110 """
1111 self._assure_config_name_safe(section, "section")
1112 self._assure_config_name_safe(option, "option")
1113 value_str = self._value_to_string_safe(value)
1114 if not self.has_section(section):
1115 self.add_section(section)
1116 super().set(section, option, value_str)
1117 return self
1119 @needs_values
1120 @set_dirty_and_flush_changes
1121 def add_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
1122 """Add a value for the given option in section.
1124 This will create the section if required, and will not throw as opposed to the
1125 default ConfigParser ``set`` method. The value becomes the new value of the
1126 option as returned by :meth:`get_value`, and appends to the list of values
1127 returned by :meth:`get_values`.
1129 :param section:
1130 Name of the section in which the option resides or should reside.
1132 :param option:
1133 Name of the option.
1135 :param value:
1136 Value to add to option. It must be a string or convertible to a string.
1138 :return:
1139 This instance
1140 """
1141 self._assure_config_name_safe(section, "section")
1142 self._assure_config_name_safe(option, "option")
1143 value_str = self._value_to_string_safe(value)
1144 if not self.has_section(section):
1145 self.add_section(section)
1146 self._sections[section].add(option, value_str)
1147 return self
1149 def rename_section(self, section: str, new_name: str) -> "GitConfigParser":
1150 """Rename the given section to `new_name`.
1152 :raise ValueError:
1153 If:
1155 * `section` doesn't exist.
1156 * A section with `new_name` does already exist.
1158 :return:
1159 This instance
1160 """
1161 if not self.has_section(section):
1162 raise ValueError("Source section '%s' doesn't exist" % section)
1163 self._assure_config_name_safe(new_name, "section")
1164 if self.has_section(new_name):
1165 raise ValueError("Destination section '%s' already exists" % new_name)
1167 super().add_section(new_name)
1168 new_section = self._sections[new_name]
1169 for k, vs in self.items_all(section):
1170 new_section.setall(k, vs)
1171 # END for each value to copy
1173 # This call writes back the changes, which is why we don't have the respective
1174 # decorator.
1175 self.remove_section(section)
1176 return self