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."""
72UNQUOTED_ESCAPES = {"b": "\b", "n": "\n", "t": "\t", '"': '"', "\\": "\\"}
73"""Escape sequences git recognizes in an unquoted value.
75Any other backslash escape is an error for git, which is why a value holding one is
76left undecoded rather than rejected.
77"""
80class MetaParserBuilder(abc.ABCMeta): # noqa: B024
81 """Utility class wrapping base-class methods into decorators that assure read-only
82 properties."""
84 def __new__(cls, name: str, bases: Tuple, clsdict: Dict[str, Any]) -> "MetaParserBuilder":
85 """Equip all base-class methods with a needs_values decorator, and all non-const
86 methods with a :func:`set_dirty_and_flush_changes` decorator in addition to
87 that.
88 """
89 kmm = "_mutating_methods_"
90 if kmm in clsdict:
91 mutating_methods = clsdict[kmm]
92 for base in bases:
93 methods = (t for t in inspect.getmembers(base, inspect.isroutine) if not t[0].startswith("_"))
94 for method_name, method in methods:
95 if method_name in clsdict:
96 continue
97 method_with_values = needs_values(method)
98 if method_name in mutating_methods:
99 method_with_values = set_dirty_and_flush_changes(method_with_values)
100 # END mutating methods handling
102 clsdict[method_name] = method_with_values
103 # END for each name/method pair
104 # END for each base
105 # END if mutating methods configuration is set
107 new_type = super().__new__(cls, name, bases, clsdict)
108 return new_type
111def needs_values(func: Callable[..., _T]) -> Callable[..., _T]:
112 """Return a method for ensuring we read values (on demand) before we try to access
113 them."""
115 @wraps(func)
116 def assure_data_present(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
117 self.read()
118 return func(self, *args, **kwargs)
120 # END wrapper method
121 return assure_data_present
124def set_dirty_and_flush_changes(non_const_func: Callable[..., _T]) -> Callable[..., _T]:
125 """Return a method that checks whether given non constant function may be called.
127 If so, the instance will be set dirty. Additionally, we flush the changes right to
128 disk.
129 """
131 def flush_changes(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
132 rval = non_const_func(self, *args, **kwargs)
133 self._dirty = True
134 self.write()
135 return rval
137 # END wrapper method
138 flush_changes.__name__ = non_const_func.__name__
139 return flush_changes
142class SectionConstraint(Generic[T_ConfigParser]):
143 """Constrains a ConfigParser to only option commands which are constrained to
144 always use the section we have been initialized with.
146 It supports all ConfigParser methods that operate on an option.
148 :note:
149 If used as a context manager, will release the wrapped ConfigParser.
150 """
152 __slots__ = ("_config", "_section_name")
154 _valid_attrs_ = (
155 "get_value",
156 "set_value",
157 "get",
158 "set",
159 "getint",
160 "getfloat",
161 "getboolean",
162 "has_option",
163 "remove_section",
164 "remove_option",
165 "options",
166 )
168 def __init__(self, config: T_ConfigParser, section: str) -> None:
169 self._config = config
170 self._section_name = section
172 def __del__(self) -> None:
173 # Yes, for some reason, we have to call it explicitly for it to work in PY3 !
174 # Apparently __del__ doesn't get call anymore if refcount becomes 0
175 # Ridiculous ... .
176 self._config.release()
178 def __getattr__(self, attr: str) -> Any:
179 if attr in self._valid_attrs_:
180 return lambda *args, **kwargs: self._call_config(attr, *args, **kwargs)
181 return super().__getattribute__(attr)
183 def _call_config(self, method: str, *args: Any, **kwargs: Any) -> Any:
184 """Call the configuration at the given method which must take a section name as
185 first argument."""
186 return getattr(self._config, method)(self._section_name, *args, **kwargs)
188 @property
189 def config(self) -> T_ConfigParser:
190 """return: ConfigParser instance we constrain"""
191 return self._config
193 def release(self) -> None:
194 """Equivalent to :meth:`GitConfigParser.release`, which is called on our
195 underlying parser instance."""
196 return self._config.release()
198 def __enter__(self) -> "SectionConstraint[T_ConfigParser]":
199 self._config.__enter__()
200 return self
202 def __exit__(self, exception_type: str, exception_value: str, traceback: str) -> None:
203 self._config.__exit__(exception_type, exception_value, traceback)
206def _normalize_name(name: str) -> str:
207 """Fold section and option names, leaving quoted subsections unchanged."""
208 prefix, separator, subsection = name.partition('"')
209 return prefix.lower() + separator + subsection
212class _OMD(OrderedDict_OMD):
213 """Ordered multi-dict matching config names while retaining their first spelling."""
215 def __init__(self, *args: Any, **kwargs: Any) -> None:
216 self._keymap: Dict[str, str] = {}
217 super().__init__(*args, **kwargs)
219 def _key(self, key: str) -> str:
220 stored = self._keymap.get(_normalize_name(key), key)
221 return stored if super().__contains__(stored) else key
223 def __contains__(self, key: object) -> bool:
224 return isinstance(key, str) and super().__contains__(self._key(key))
226 def __delitem__(self, key: str) -> None:
227 super().__delitem__(self._key(key))
228 del self._keymap[_normalize_name(key)]
230 def __setitem__(self, key: str, value: _T) -> None:
231 self.setall(key, [value])
233 def clear(self) -> None:
234 super().clear()
235 self._keymap.clear()
237 def add(self, key: str, value: Any) -> None:
238 if key not in self:
239 self[key] = value
240 return
242 self.getall(key).append(value)
244 def setall(self, key: str, values: List[_T]) -> None:
245 key = self._key(key)
246 super().__setitem__(key, values)
247 self._keymap[_normalize_name(key)] = key
249 def __getitem__(self, key: str) -> Any:
250 return super().__getitem__(self._key(key))[-1]
252 def getlast(self, key: str) -> Any:
253 return self[key]
255 def setlast(self, key: str, value: Any) -> None:
256 if key not in self:
257 self[key] = value
258 return
260 self.getall(key)[-1] = value
262 def get(self, key: str, default: Union[_T, None] = None) -> Union[_T, None]: # type: ignore[override]
263 return super().get(self._key(key), [default])[-1]
265 def getall(self, key: str) -> List[_T]:
266 return super().__getitem__(self._key(key))
268 def items(self) -> List[Tuple[str, _T]]: # type: ignore[override]
269 """List of (key, last value for key)."""
270 return [(k, self[k]) for k in self]
272 def items_all(self) -> List[Tuple[str, List[_T]]]:
273 """List of (key, list of values for key)."""
274 return [(k, self.getall(k)) for k in self]
277def get_config_path(config_level: Lit_config_levels) -> str:
278 # We do not support an absolute path of the gitconfig on Windows.
279 # Use the global config instead.
280 if sys.platform == "win32" and config_level == "system":
281 config_level = "global"
283 if config_level == "system":
284 return "/etc/gitconfig"
285 elif config_level == "user":
286 config_home = os.environ.get("XDG_CONFIG_HOME") or osp.join(os.environ.get("HOME", "~"), ".config")
287 return osp.normpath(osp.expanduser(osp.join(config_home, "git", "config")))
288 elif config_level == "global":
289 return osp.normpath(osp.expanduser("~/.gitconfig"))
290 elif config_level == "repository":
291 raise ValueError("No repo to get repository configuration from. Use Repo._get_config_path")
292 else:
293 # Should not reach here. Will raise ValueError if does. Static typing will warn
294 # about missing elifs.
295 assert_never( # type: ignore[unreachable]
296 config_level,
297 ValueError(f"Invalid configuration level: {config_level!r}"),
298 )
301class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder):
302 """Implements specifics required to read git style configuration files.
304 This variation behaves much like the :manpage:`git-config(1)` command, such that the
305 configuration will be read on demand based on the filepath given during
306 initialization.
308 The changes will automatically be written once the instance goes out of scope, but
309 can be triggered manually as well.
311 The configuration file will be locked if you intend to change values preventing
312 other instances to write concurrently.
314 :note:
315 Section and option names are case-insensitive; quoted subsection names are
316 case-sensitive. Names retain their first spelling when enumerated or written.
317 Case variants are merged, preserving all values in the order they are read.
319 :note:
320 If used as a context manager, this will release the locked file.
322 :note:
323 Options without a value are stored as ``None`` and written without ``=``.
324 :meth:`get_value` and :meth:`get_values` return an empty string for them,
325 while :meth:`getboolean` returns ``True``. An explicit empty value is
326 stored as an empty string and reads as ``False`` with :meth:`getboolean`.
327 """
329 # { Configuration
330 t_lock = LockFile
331 """The lock type determines the type of lock to use in new configuration readers.
333 They must be compatible to the :class:`~git.util.LockFile` interface.
334 A suitable alternative would be the :class:`~git.util.BlockingLockFile`.
335 """
337 re_comment = re.compile(r"^\s*[#;]")
338 # } END configuration
340 optvalueonly_source = r"\s*(?P<option>[^:=\s#;][^:=#;]*)"
342 OPTVALUEONLY = re.compile(optvalueonly_source)
344 # The option name class [^:=#;]* already consumes any spaces up to the ":" or "=",
345 # so a second \s* before the indicator would overlap it and backtrack quadratically
346 # on a line that never reaches an indicator (for example a key followed by a long
347 # whitespace run). Drop the redundant \s*; the name is right-stripped after parsing.
348 OPTCRE = re.compile(optvalueonly_source + r"(?P<vi>[:=])\s*" + r"(?P<value>.*)$")
350 del optvalueonly_source
352 _mutating_methods_ = ("add_section", "remove_section", "remove_option", "set")
353 """Names of :class:`~configparser.RawConfigParser` methods able to change the
354 instance."""
356 def __init__(
357 self,
358 file_or_files: Union[None, PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = None,
359 read_only: bool = True,
360 merge_includes: bool = True,
361 config_level: Union[Lit_config_levels, None] = None,
362 repo: Union["Repo", None] = None,
363 ) -> None:
364 """Initialize a configuration reader to read the given `file_or_files` and to
365 possibly allow changes to it by setting `read_only` False.
367 :param file_or_files:
368 A file path or file object, or a sequence of possibly more than one of them.
370 :param read_only:
371 If ``True``, the ConfigParser may only read the data, but not change it.
372 If ``False``, only a single file path or file object may be given. We will
373 write back the changes when they happen, or when the ConfigParser is
374 released. This will not happen if other configuration files have been
375 included.
377 :param merge_includes:
378 If ``True``, we will read files mentioned in ``[include]`` sections and
379 merge their contents into ours. This makes it impossible to write back an
380 individual configuration file. Thus, if you want to modify a single
381 configuration file, turn this off to leave the original dataset unaltered
382 when reading it.
384 :param repo:
385 Reference to repository to use if ``[includeIf]`` sections are found in
386 configuration files.
387 """
388 cp.RawConfigParser.__init__(self, dict_type=cast(Any, _OMD), allow_no_value=True)
389 self._dict: Callable[..., _OMD]
390 self._defaults: _OMD
391 self._sections: _OMD
393 # Used in Python 3. Needs to stay in sync with sections for underlying
394 # implementation to work.
395 if not hasattr(self, "_proxies"):
396 self._proxies = self._dict()
398 if file_or_files is not None:
399 self._file_or_files: Union[PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = file_or_files
400 else:
401 if config_level is None:
402 if read_only:
403 self._file_or_files = [
404 get_config_path(cast(Lit_config_levels, f)) for f in CONFIG_LEVELS if f != "repository"
405 ]
406 else:
407 raise ValueError("No configuration level or configuration files specified")
408 else:
409 self._file_or_files = [get_config_path(config_level)]
411 self._read_only = read_only
412 self._dirty = False
413 self._is_initialized = False
414 self._merge_includes = merge_includes
415 self._repo = repo
416 self._lock: Union["LockFile", None] = None
417 self._acquire_lock()
419 def _acquire_lock(self) -> None:
420 if not self._read_only:
421 if not self._lock:
422 if isinstance(self._file_or_files, (str, os.PathLike)):
423 file_or_files = self._file_or_files
424 elif isinstance(self._file_or_files, (tuple, list, Sequence)):
425 raise ValueError(
426 "Write-ConfigParsers can operate on a single file only, multiple files have been passed"
427 )
428 else:
429 file_or_files = self._file_or_files.name
431 # END get filename from handle/stream
432 # Initialize lock base - we want to write.
433 self._lock = self.t_lock(file_or_files)
434 # END lock check
436 self._lock._obtain_lock()
437 # END read-only check
439 def __del__(self) -> None:
440 """Write pending changes if required and release locks."""
441 # NOTE: Only consistent in Python 2.
442 self.release()
444 def __enter__(self) -> "GitConfigParser":
445 self._acquire_lock()
446 return self
448 def __exit__(self, *args: Any) -> None:
449 self.release()
451 def release(self) -> None:
452 """Flush changes and release the configuration write lock. This instance must
453 not be used anymore afterwards.
455 In Python 3, it's required to explicitly release locks and flush changes, as
456 ``__del__`` is not called deterministically anymore.
457 """
458 # Checking for the lock here makes sure we do not raise during write()
459 # in case an invalid parser was created who could not get a lock.
460 if self.read_only or (self._lock and not self._lock._has_lock()):
461 return
463 try:
464 self.write()
465 except OSError:
466 _logger.error("Exception during destruction of GitConfigParser", exc_info=True)
467 except ReferenceError:
468 # This happens in Python 3... and usually means that some state cannot be
469 # written as the sections dict cannot be iterated. This usually happens when
470 # the interpreter is shutting down. Can it be fixed?
471 pass
472 finally:
473 if self._lock is not None:
474 self._lock._release_lock()
476 def optionxform(self, optionstr: str) -> str:
477 """Do not transform options in any way when writing."""
478 return optionstr
480 def _read(self, fp: Union[BufferedReader, IO[bytes]], fpname: str) -> None:
481 """Originally a direct copy of the Python 2.4 version of
482 :meth:`RawConfigParser._read <configparser.RawConfigParser._read>`, to ensure it
483 uses ordered dicts.
485 The ordering bug was fixed in Python 2.4, and dict itself keeps ordering since
486 Python 3.7. This has some other changes, especially that it ignores initial
487 whitespace, since git uses tabs. (Big comments are removed to be more compact.)
488 """
489 cursect = None # None, or a dictionary.
490 optname = None
491 lineno = 0
492 is_multi_line = False
493 e = None # None, or an exception.
495 def string_decode(v: str) -> str:
496 if v and v.endswith("\\"):
497 v = v[:-1]
498 # END cut trailing escapes to prevent decode error
500 escapes = {"b": "\b", "n": "\n", "r": "\r", "t": "\t", '"': '"', "\\": "\\"}
501 return re.sub(r"\\(.)", lambda match: escapes.get(match.group(1), match.group(0)), v)
503 # END string_decode
505 def is_line_continuation(value: str) -> bool:
506 quoted = escaped = False
507 for char in value:
508 if escaped:
509 escaped = False
510 elif char == "\\":
511 escaped = True
512 elif char == '"':
513 quoted = not quoted
514 elif char in "#;" and not quoted:
515 return False
516 return escaped
518 def strip_inline_comment(value: str) -> Tuple[str, bool]:
519 """Cut an unquoted ``#`` or ``;`` comment and report whether a quote is open.
521 Quoting and backslash escapes are honoured, so a ``#`` inside a quoted
522 value is literal and an unterminated quote swallows the rest of the line.
523 """
524 quoted = escaped = False
525 for index, char in enumerate(value):
526 if escaped:
527 escaped = False
528 elif char == "\\":
529 escaped = True
530 elif char == '"':
531 quoted = not quoted
532 elif char in "#;" and not quoted:
533 return value[:index], False
534 return value, quoted
536 def parse_value(value: str) -> str:
537 parsed: List[str] = []
538 whitespace: List[str] = []
539 quoted = escaped = False
540 for char in value:
541 if escaped:
542 parsed.append(UNQUOTED_ESCAPES.get(char, "\\" + char))
543 escaped = False
544 continue
545 if char.isspace() and not quoted:
546 if parsed:
547 whitespace.append(char)
548 continue
549 if char in "#;" and not quoted:
550 break
551 parsed.extend(whitespace)
552 whitespace.clear()
553 if char == "\\":
554 escaped = True
555 elif char == '"':
556 quoted = not quoted
557 else:
558 parsed.append(char)
559 return "".join(parsed)
561 def escapes_are_known(value: str) -> bool:
562 """Whether every backslash in `value` introduces an escape git recognises.
564 Git rejects an unquoted value carrying an unknown escape outright. GitPython
565 tolerates such values, because that is what lets a Windows path such as
566 ``C:\\Users\\me`` stay literal, so a value is decoded only when every one
567 of its backslashes is part of a known escape. Decoding part of a value would
568 otherwise turn the ``\\t`` of ``\\Temp\\test`` into a tab while leaving the
569 ``\\U`` of ``\\Users`` alone.
570 """
571 index = value.find("\\")
572 while index >= 0:
573 # A trailing backslash has nothing to escape and git drops it, so
574 # only a backslash followed by an unknown escape disqualifies the
575 # value from being decoded.
576 if index + 1 < len(value) and value[index + 1] not in UNQUOTED_ESCAPES:
577 return False
578 index = value.find("\\", index + 2)
579 return True
581 while True:
582 # We assume to read binary!
583 raw_line = fp.readline()
584 if not raw_line:
585 break
586 if lineno == 0 and raw_line.startswith(b"\xef\xbb\xbf"):
587 # A UTF-8 BOM is not part of the content. git skips it, so a
588 # config file written by a Windows editor still parses.
589 raw_line = raw_line[3:]
590 line = raw_line.decode(defenc)
591 lineno = lineno + 1
592 # Comment or blank line?
593 if line.strip() == "" or self.re_comment.match(line):
594 continue
595 if line.split(None, 1)[0].lower() == "rem" and line[0] in "rR":
596 # No leading whitespace.
597 continue
599 # Is it a section header?
600 mo = self.SECTCRE.match(line.strip())
601 if not is_multi_line and mo:
602 sectname: str = mo.group("header").strip()
603 if sectname in self._sections:
604 cursect = self._sections[sectname]
605 elif sectname == cp.DEFAULTSECT:
606 cursect = self._defaults
607 else:
608 cursect = self._dict((("__name__", sectname),))
609 self._sections[sectname] = cursect
610 self._proxies[sectname] = None
611 # So sections can't start with a continuation line.
612 optname = None
613 # No section header in the file?
614 elif cursect is None:
615 raise cp.MissingSectionHeaderError(fpname, lineno, line)
616 # An option line?
617 elif not is_multi_line:
618 mo = self.OPTCRE.match(line)
619 if mo:
620 # We might just have handled the last line, which could contain a quotation we want to remove.
621 optname, vi, optval = mo.group("option", "vi", "value")
622 optname = self.optionxform(optname.rstrip())
624 optval, quote_open = strip_inline_comment(optval)
625 optval = optval.strip()
627 if len(optval) < 2 or optval[0] != '"':
628 # Does not open quoting.
629 # A value ending in an odd number of backslashes
630 # continues on the next line, exactly as git does: the
631 # final backslash and the newline are removed and the
632 # next line is appended before the complete value is
633 # parsed. An even number means the last backslash is
634 # escaped and the value ends there.
635 eof_continuation = False
636 while True:
637 if not is_line_continuation(optval):
638 break
639 continuation = fp.readline()
640 if not continuation:
641 # Keep the backslash until parsing so preceding
642 # whitespace is significant even though git drops it.
643 eof_continuation = True
644 break
645 lineno = lineno + 1
646 joined = continuation.decode(defenc)
647 while joined.endswith("\n") or joined.endswith("\r"):
648 joined = joined[:-1]
649 optval = optval[:-1] + joined
650 # Git decodes the standard escape sequences in unquoted
651 # values, but only a value in which every backslash is part
652 # of a known escape is decoded, so that a Windows path keeps
653 # its backslashes instead of losing them one escape at a time.
654 optval, _ = strip_inline_comment(optval)
655 if escapes_are_known(optval):
656 optval = parse_value(optval)
657 elif eof_continuation:
658 optval = optval[:-1]
659 elif quote_open:
660 # Opens quoting and does not close: appears to start multi-line quoting.
661 is_multi_line = True
662 optval = string_decode(optval[1:])
663 elif re.search(r'(?:^|[^\\])(?:\\\\)*"', optval[1:-1]):
664 # Preserve values containing additional unescaped quotes.
665 pass
666 else:
667 # Opens and closes quoting.
668 optval = string_decode(optval[1:-1])
670 # Preserves multiple values for duplicate optnames.
671 cursect.add(optname, optval)
672 else:
673 # A valueless option is an implicit boolean true, not an empty value.
674 mo = self.OPTVALUEONLY.fullmatch(line)
675 if mo:
676 optname = self.optionxform(mo.group("option").rstrip())
677 cursect.add(optname, None)
678 else:
679 if not e:
680 e = cp.ParsingError(fpname)
681 e.append(lineno, repr(line))
682 continue
683 else:
684 line = line.rstrip()
685 if line.endswith('"'):
686 is_multi_line = False
687 line = line[:-1]
688 # END handle quotations
689 optval = cursect.getlast(optname)
690 cursect.setlast(optname, optval + string_decode(line))
691 # END parse section or option
692 # END while reading
694 # If any parsing errors occurred, raise an exception.
695 if e:
696 raise e
698 def _has_includes(self) -> Union[bool, int]:
699 return self._merge_includes and len(self._included_paths())
701 def _included_paths(self) -> List[Tuple[str, str]]:
702 """List all paths that must be included to configuration.
704 :return:
705 The list of paths, where each path is a tuple of (option, value).
706 """
708 def _all_items(section: str) -> List[Tuple[str, str]]:
709 """Return all (key, value) pairs for a section, including duplicate keys."""
710 return [
711 (key, value)
712 for key, values in self._sections[section].items_all()
713 if key != "__name__"
714 for value in values
715 if value is not None
716 ]
718 paths = []
720 for section in self.sections():
721 normalized_section = _normalize_name(section)
722 if normalized_section == "include":
723 paths += _all_items(section)
725 match = CONDITIONAL_INCLUDE_REGEXP.search(normalized_section)
726 if match is None or self._repo is None:
727 continue
729 keyword = match.group(1)
730 value = match.group(2).strip()
732 if keyword in ["gitdir", "gitdir/i"]:
733 value = osp.expanduser(value)
734 git_dir = os.fspath(self._repo.git_dir) if self._repo.git_dir else None
735 if sys.platform == "win32":
736 git_dir = git_dir.replace("\\", "/") if git_dir else None
738 drive, _tail = osp.splitdrive(value)
739 if not drive and not any(value.startswith(s) for s in ["./", "/"]):
740 value = "**/" + value
741 if value.endswith("/"):
742 value += "**"
744 # Ensure that glob is always case insensitive if required.
745 if keyword.endswith("/i"):
746 value = re.sub(
747 r"[a-zA-Z]",
748 lambda m: f"[{m.group().lower()!r}{m.group().upper()!r}]",
749 value,
750 )
751 if git_dir and fnmatch.fnmatchcase(git_dir, value):
752 paths += _all_items(section)
754 elif keyword == "onbranch":
755 try:
756 branch_name = self._repo.active_branch.name
757 except TypeError:
758 # Ignore section if active branch cannot be retrieved.
759 continue
761 if fnmatch.fnmatchcase(branch_name, value):
762 paths += _all_items(section)
763 elif keyword == "hasconfig:remote.*.url":
764 for remote in self._repo.remotes:
765 if fnmatch.fnmatchcase(remote.url, value):
766 paths += _all_items(section)
767 break
768 return paths
770 def read(self) -> None: # type: ignore[override]
771 """Read the data stored in the files we have been initialized with.
773 This will ignore files that cannot be read, possibly leaving an empty
774 configuration.
776 :raise IOError:
777 If a file cannot be handled.
778 """
779 if self._is_initialized:
780 return
781 self._is_initialized = True
783 files_to_read: List[Union[PathLike, IO]] = [""]
784 if isinstance(self._file_or_files, (str, os.PathLike)):
785 # For str or Path, as str is a type of Sequence.
786 files_to_read = [self._file_or_files]
787 elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
788 # Could merge with above isinstance once runtime type known.
789 files_to_read = [self._file_or_files]
790 else: # For lists or tuples.
791 files_to_read = list(self._file_or_files)
792 # END ensure we have a copy of the paths to handle
794 files_to_read = [osp.abspath(path) if isinstance(path, (str, os.PathLike)) else path for path in files_to_read]
796 seen = set(files_to_read)
797 num_read_include_files = 0
798 while files_to_read:
799 file_path = files_to_read.pop(0)
800 file_ok = False
802 if hasattr(file_path, "seek"):
803 # Must be a file-object.
804 # TODO: Replace cast with assert to narrow type, once sure.
805 file_path = cast(IO[bytes], file_path)
806 self._read(file_path, file_path.name)
807 else:
808 try:
809 with open(file_path, "rb") as fp:
810 file_ok = True
811 self._read(fp, fp.name)
812 except OSError:
813 continue
815 # Read includes and append those that we didn't handle yet. We expect all
816 # paths to be normalized and absolute (and will ensure that is the case).
817 if self._has_includes():
818 for _, include_path in self._included_paths():
819 if include_path.startswith("~"):
820 include_path = osp.expanduser(include_path)
821 if not osp.isabs(include_path):
822 if not file_ok:
823 continue
824 # END ignore relative paths if we don't know the configuration file path
825 file_path = cast(PathLike, file_path)
826 assert osp.isabs(file_path), "Need absolute paths to be sure our cycle checks will work"
827 include_path = osp.join(osp.dirname(file_path), include_path)
828 # END make include path absolute
829 include_path = osp.normpath(include_path)
830 if include_path in seen or not os.access(include_path, os.R_OK):
831 continue
832 seen.add(include_path)
833 # Insert included file to the top to be considered first.
834 files_to_read.insert(0, include_path)
835 num_read_include_files += 1
836 # END each include path in configuration file
837 # END handle includes
838 # END for each file object to read
840 # If there was no file included, we can safely write back (potentially) the
841 # configuration file without altering its meaning.
842 if num_read_include_files == 0:
843 self._merge_includes = False
845 def _write(self, fp: IO) -> None:
846 """Write an .ini-format representation of the configuration state in
847 git compatible format."""
849 def write_section(name: str, section_dict: _OMD) -> None:
850 fp.write(("[%s]\n" % name).encode(defenc))
852 values: List[Any]
853 v: Any
854 for key, values in section_dict.items_all():
855 if key == "__name__":
856 continue
858 for v in values:
859 if v is None:
860 fp.write(("\t%s\n" % key).encode(defenc))
861 continue
862 value = self._value_to_string(v)
863 if any(char in value for char in '\n\t\b\\"#;') or value[:1].isspace() or value[-1:].isspace():
864 value = value.replace("\\", "\\\\").replace('"', '\\"')
865 value = '"%s\\\n"' % value.replace("\n", "\\n").replace("\t", "\\t").replace("\b", "\\b")
866 fp.write(("\t%s = %s\n" % (key, value)).encode(defenc))
867 # END if key is not __name__
869 # END section writing
871 if self._defaults:
872 write_section(cp.DEFAULTSECT, self._defaults)
873 value: _OMD
875 for name, value in self._sections.items():
876 write_section(name, value)
878 def items(self, section_name: str) -> List[Tuple[str, Union[str, None]]]: # type: ignore[override]
879 """:return: list((option, value), ...) pairs of all items in the given section"""
880 return [(k, v) for k, v in super().items(section_name) if k != "__name__"]
882 def items_all(self, section_name: str) -> List[Tuple[str, List[Union[str, None]]]]:
883 """:return: list((option, [values...]), ...) pairs of all items in the given section"""
884 rv = _OMD(self._defaults)
886 for k, vs in self._sections[section_name].items_all():
887 if k == "__name__":
888 continue
890 if k in rv and rv.getall(k) == vs:
891 continue
893 for v in vs:
894 rv.add(k, v)
896 return rv.items_all()
898 @needs_values
899 def write(self) -> None:
900 """Write changes to our file, if there are changes at all.
902 :raise IOError:
903 If this is a read-only writer instance or if we could not obtain a file
904 lock.
905 """
906 self._assure_writable("write")
907 if not self._dirty:
908 return
910 if isinstance(self._file_or_files, (list, tuple)):
911 raise AssertionError(
912 "Cannot write back if there is not exactly a single file to write to, have %i files"
913 % len(self._file_or_files)
914 )
915 # END assert multiple files
917 if self._has_includes():
918 _logger.debug(
919 "Skipping write-back of configuration file as include files were merged in."
920 + "Set merge_includes=False to prevent this."
921 )
922 return
923 # END stop if we have include files
925 sections: List[_OMD] = [self._defaults]
926 section: _OMD
927 stored_section: _OMD
928 values: List[Any]
929 raw_value: Any
930 for _, stored_section in self._sections.items():
931 sections.append(stored_section)
932 for section in sections:
933 for key, values in section.items_all():
934 if key != "__name__":
935 for raw_value in values:
936 if raw_value is None:
937 continue
938 if "\r" in self._value_to_string(raw_value) or "\x00" in self._value_to_string(raw_value):
939 raise ValueError("Git config values must not contain CR or NUL")
941 fp = self._file_or_files
943 # We have a physical file on disk, so get a lock.
944 is_file_lock = isinstance(fp, (str, os.PathLike, IOBase)) # TODO: Use PathLike (having dropped 3.5).
945 if is_file_lock and self._lock is not None: # Else raise error?
946 self._lock._obtain_lock()
948 if not hasattr(fp, "seek"):
949 fp = cast(PathLike, fp)
950 with open(fp, "wb") as fp_open:
951 self._write(fp_open)
952 else:
953 fp = cast("BytesIO", fp)
954 fp.seek(0)
955 # Make sure we do not overwrite into an existing file.
956 if hasattr(fp, "truncate"):
957 fp.truncate()
958 self._write(fp)
960 def _assure_writable(self, method_name: str) -> None:
961 if self.read_only:
962 raise OSError(f"Cannot execute non-constant method {self}.{method_name}")
964 def add_section(self, section: "cp._SectionName") -> None:
965 """Assures added options will stay in order."""
966 self._assure_config_name_safe(section, "section")
967 return super().add_section(section)
969 @property
970 def read_only(self) -> bool:
971 """:return: ``True`` if this instance may change the configuration file"""
972 return self._read_only
974 def get_value(
975 self,
976 section: str,
977 option: str,
978 default: Union[int, float, str, bool, None] = None,
979 ) -> Union[int, float, str, bool]:
980 """Get an option's value.
982 If multiple values are specified for this option in the section, the last one
983 specified is returned.
985 :param default:
986 If not ``None``, the given default value will be returned in case the option
987 did not exist.
989 :return:
990 A properly typed value, either int, float, string or bool
992 :raise TypeError:
993 In case the value could not be understood.
994 Otherwise the exceptions known to the ConfigParser will be raised.
995 """
996 try:
997 valuestr = self.get(section, option)
998 except Exception:
999 if default is not None:
1000 return default
1001 raise
1003 return self._string_to_value(valuestr)
1005 def get_values(
1006 self,
1007 section: str,
1008 option: str,
1009 default: Union[int, float, str, bool, None] = None,
1010 ) -> List[Union[int, float, str, bool]]:
1011 """Get an option's values.
1013 If multiple values are specified for this option in the section, all are
1014 returned.
1016 :param default:
1017 If not ``None``, a list containing the given default value will be returned
1018 in case the option did not exist.
1020 :return:
1021 A list of properly typed values, either int, float, string or bool
1023 :raise TypeError:
1024 In case the value could not be understood.
1025 Otherwise the exceptions known to the ConfigParser will be raised.
1026 """
1027 try:
1028 self.sections()
1029 lst = self._sections[section].getall(option)
1030 except Exception:
1031 if default is not None:
1032 return [default]
1033 raise
1035 return [self._string_to_value(valuestr) for valuestr in lst]
1037 def _convert_to_boolean(self, value: Union[str, None]) -> bool:
1038 if value is None:
1039 return True
1040 if value == "":
1041 return False
1042 try:
1043 return self.BOOLEAN_STATES[value.lower()]
1044 except KeyError:
1045 raise ValueError("Not a boolean: %s" % value) from None
1047 def _string_to_value(self, valuestr: Union[str, None]) -> Union[int, float, str, bool]:
1048 if valuestr is None:
1049 return ""
1050 types = (int, float)
1051 for numtype in types:
1052 try:
1053 val = numtype(valuestr)
1054 # Reject NaN without coercing exact integer values to float.
1055 if numtype is float and val != float(valuestr):
1056 continue
1057 return val
1058 except (ValueError, TypeError):
1059 continue
1060 # END for each numeric type
1062 # Try boolean values as git uses them.
1063 vl = valuestr.lower()
1064 if vl in ("false", "no", "off"):
1065 return False
1066 if vl in ("true", "yes", "on"):
1067 return True
1069 if not isinstance(valuestr, str):
1070 raise TypeError(
1071 "Invalid value type: only int, long, float and str are allowed",
1072 valuestr,
1073 )
1075 return valuestr
1077 def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str:
1078 if isinstance(value, (int, float, bool)):
1079 return str(value)
1080 return force_text(value)
1082 def _value_to_string_safe(self, value: Union[str, bytes, int, float, bool]) -> str:
1083 value_str = self._value_to_string(value)
1084 if UNSAFE_CONFIG_CHARS_RE.search(value_str):
1085 raise ValueError("Git config values must not contain CR, LF, or NUL")
1086 return value_str
1088 def _assure_config_name_safe(self, name: "cp._SectionName", label: str) -> None:
1089 if isinstance(name, str) and UNSAFE_CONFIG_CHARS_RE.search(name):
1090 raise ValueError("Git config %s names must not contain CR, LF, or NUL" % label)
1091 if label == "option" and isinstance(name, str) and not VALID_CONFIG_OPTION_NAME_RE.fullmatch(name):
1092 raise ValueError("Git config option names may contain only letters, digits, '-', '_', or '.'")
1093 if label == "section" and isinstance(name, str):
1094 in_quotes = False
1095 escaped = False
1096 for index, char in enumerate(name):
1097 if escaped:
1098 escaped = False
1099 elif in_quotes and char == "\\":
1100 escaped = True
1101 elif char == '"':
1102 if not in_quotes and (index == 0 or name[index - 1] not in " \t"):
1103 raise ValueError("Git config quoted subsection names must begin after whitespace")
1104 in_quotes = not in_quotes
1105 elif char == "]" and not in_quotes:
1106 raise ValueError("Git config section names must not contain an unquoted closing bracket")
1107 if in_quotes:
1108 raise ValueError("Git config section names must not contain an unterminated quote")
1110 @needs_values
1111 @set_dirty_and_flush_changes
1112 def set(
1113 self,
1114 section: str,
1115 option: str,
1116 value: Union[str, bytes, int, float, bool, None] = None,
1117 ) -> None:
1118 self._assure_config_name_safe(section, "section")
1119 self._assure_config_name_safe(option, "option")
1120 if value is not None:
1121 value = self._value_to_string_safe(value)
1122 return super().set(section, option, value)
1124 @needs_values
1125 @set_dirty_and_flush_changes
1126 def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
1127 """Set the given option in section to the given value.
1129 This will create the section if required, and will not throw as opposed to the
1130 default ConfigParser ``set`` method.
1132 :param section:
1133 Name of the section in which the option resides or should reside.
1135 :param option:
1136 Name of the options whose value to set.
1138 :param value:
1139 Value to set the option to. It must be a string or convertible to a string.
1141 :return:
1142 This instance
1143 """
1144 self._assure_config_name_safe(section, "section")
1145 self._assure_config_name_safe(option, "option")
1146 value_str = self._value_to_string_safe(value)
1147 if not self.has_section(section):
1148 self.add_section(section)
1149 super().set(section, option, value_str)
1150 return self
1152 @needs_values
1153 @set_dirty_and_flush_changes
1154 def add_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
1155 """Add a value for the given option in section.
1157 This will create the section if required, and will not throw as opposed to the
1158 default ConfigParser ``set`` method. The value becomes the new value of the
1159 option as returned by :meth:`get_value`, and appends to the list of values
1160 returned by :meth:`get_values`.
1162 :param section:
1163 Name of the section in which the option resides or should reside.
1165 :param option:
1166 Name of the option.
1168 :param value:
1169 Value to add to option. It must be a string or convertible to a string.
1171 :return:
1172 This instance
1173 """
1174 self._assure_config_name_safe(section, "section")
1175 self._assure_config_name_safe(option, "option")
1176 value_str = self._value_to_string_safe(value)
1177 if not self.has_section(section):
1178 self.add_section(section)
1179 self._sections[section].add(option, value_str)
1180 return self
1182 def rename_section(self, section: str, new_name: str) -> "GitConfigParser":
1183 """Rename the given section to `new_name`.
1185 :raise ValueError:
1186 If:
1188 * `section` doesn't exist.
1189 * A section with `new_name` does already exist.
1191 :return:
1192 This instance
1193 """
1194 if not self.has_section(section):
1195 raise ValueError("Source section '%s' doesn't exist" % section)
1196 self._assure_config_name_safe(new_name, "section")
1197 if self.has_section(new_name):
1198 raise ValueError("Destination section '%s' already exists" % new_name)
1200 super().add_section(new_name)
1201 new_section = self._sections[new_name]
1202 for k, vs in self.items_all(section):
1203 new_section.setall(k, vs)
1204 # END for each value to copy
1206 # This call writes back the changes, which is why we don't have the respective
1207 # decorator.
1208 self.remove_section(section)
1209 return self