Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/config.py: 72%
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
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"""
76class MetaParserBuilder(abc.ABCMeta): # noqa: B024
77 """Utility class wrapping base-class methods into decorators that assure read-only
78 properties."""
80 def __new__(cls, name: str, bases: Tuple, clsdict: Dict[str, Any]) -> "MetaParserBuilder":
81 """Equip all base-class methods with a needs_values decorator, and all non-const
82 methods with a :func:`set_dirty_and_flush_changes` decorator in addition to
83 that.
84 """
85 kmm = "_mutating_methods_"
86 if kmm in clsdict:
87 mutating_methods = clsdict[kmm]
88 for base in bases:
89 methods = (t for t in inspect.getmembers(base, inspect.isroutine) if not t[0].startswith("_"))
90 for method_name, method in methods:
91 if method_name in clsdict:
92 continue
93 method_with_values = needs_values(method)
94 if method_name in mutating_methods:
95 method_with_values = set_dirty_and_flush_changes(method_with_values)
96 # END mutating methods handling
98 clsdict[method_name] = method_with_values
99 # END for each name/method pair
100 # END for each base
101 # END if mutating methods configuration is set
103 new_type = super().__new__(cls, name, bases, clsdict)
104 return new_type
107def needs_values(func: Callable[..., _T]) -> Callable[..., _T]:
108 """Return a method for ensuring we read values (on demand) before we try to access
109 them."""
111 @wraps(func)
112 def assure_data_present(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
113 self.read()
114 return func(self, *args, **kwargs)
116 # END wrapper method
117 return assure_data_present
120def set_dirty_and_flush_changes(non_const_func: Callable[..., _T]) -> Callable[..., _T]:
121 """Return a method that checks whether given non constant function may be called.
123 If so, the instance will be set dirty. Additionally, we flush the changes right to
124 disk.
125 """
127 def flush_changes(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:
128 rval = non_const_func(self, *args, **kwargs)
129 self._dirty = True
130 self.write()
131 return rval
133 # END wrapper method
134 flush_changes.__name__ = non_const_func.__name__
135 return flush_changes
138class SectionConstraint(Generic[T_ConfigParser]):
139 """Constrains a ConfigParser to only option commands which are constrained to
140 always use the section we have been initialized with.
142 It supports all ConfigParser methods that operate on an option.
144 :note:
145 If used as a context manager, will release the wrapped ConfigParser.
146 """
148 __slots__ = ("_config", "_section_name")
150 _valid_attrs_ = (
151 "get_value",
152 "set_value",
153 "get",
154 "set",
155 "getint",
156 "getfloat",
157 "getboolean",
158 "has_option",
159 "remove_section",
160 "remove_option",
161 "options",
162 )
164 def __init__(self, config: T_ConfigParser, section: str) -> None:
165 self._config = config
166 self._section_name = section
168 def __del__(self) -> None:
169 # Yes, for some reason, we have to call it explicitly for it to work in PY3 !
170 # Apparently __del__ doesn't get call anymore if refcount becomes 0
171 # Ridiculous ... .
172 self._config.release()
174 def __getattr__(self, attr: str) -> Any:
175 if attr in self._valid_attrs_:
176 return lambda *args, **kwargs: self._call_config(attr, *args, **kwargs)
177 return super().__getattribute__(attr)
179 def _call_config(self, method: str, *args: Any, **kwargs: Any) -> Any:
180 """Call the configuration at the given method which must take a section name as
181 first argument."""
182 return getattr(self._config, method)(self._section_name, *args, **kwargs)
184 @property
185 def config(self) -> T_ConfigParser:
186 """return: ConfigParser instance we constrain"""
187 return self._config
189 def release(self) -> None:
190 """Equivalent to :meth:`GitConfigParser.release`, which is called on our
191 underlying parser instance."""
192 return self._config.release()
194 def __enter__(self) -> "SectionConstraint[T_ConfigParser]":
195 self._config.__enter__()
196 return self
198 def __exit__(self, exception_type: str, exception_value: str, traceback: str) -> None:
199 self._config.__exit__(exception_type, exception_value, traceback)
202class _OMD(OrderedDict_OMD):
203 """Ordered multi-dict."""
205 def __setitem__(self, key: str, value: _T) -> None:
206 super().__setitem__(key, [value])
208 def add(self, key: str, value: Any) -> None:
209 if key not in self:
210 super().__setitem__(key, [value])
211 return
213 super().__getitem__(key).append(value)
215 def setall(self, key: str, values: List[_T]) -> None:
216 super().__setitem__(key, values)
218 def __getitem__(self, key: str) -> Any:
219 return super().__getitem__(key)[-1]
221 def getlast(self, key: str) -> Any:
222 return super().__getitem__(key)[-1]
224 def setlast(self, key: str, value: Any) -> None:
225 if key not in self:
226 super().__setitem__(key, [value])
227 return
229 prior = super().__getitem__(key)
230 prior[-1] = value
232 def get(self, key: str, default: Union[_T, None] = None) -> Union[_T, None]:
233 return super().get(key, [default])[-1]
235 def getall(self, key: str) -> List[_T]:
236 return super().__getitem__(key)
238 def items(self) -> List[Tuple[str, _T]]: # type: ignore[override]
239 """List of (key, last value for key)."""
240 return [(k, self[k]) for k in self]
242 def items_all(self) -> List[Tuple[str, List[_T]]]:
243 """List of (key, list of values for key)."""
244 return [(k, self.getall(k)) for k in self]
247def get_config_path(config_level: Lit_config_levels) -> str:
248 # We do not support an absolute path of the gitconfig on Windows.
249 # Use the global config instead.
250 if sys.platform == "win32" and config_level == "system":
251 config_level = "global"
253 if config_level == "system":
254 return "/etc/gitconfig"
255 elif config_level == "user":
256 config_home = os.environ.get("XDG_CONFIG_HOME") or osp.join(os.environ.get("HOME", "~"), ".config")
257 return osp.normpath(osp.expanduser(osp.join(config_home, "git", "config")))
258 elif config_level == "global":
259 return osp.normpath(osp.expanduser("~/.gitconfig"))
260 elif config_level == "repository":
261 raise ValueError("No repo to get repository configuration from. Use Repo._get_config_path")
262 else:
263 # Should not reach here. Will raise ValueError if does. Static typing will warn
264 # about missing elifs.
265 assert_never( # type: ignore[unreachable]
266 config_level,
267 ValueError(f"Invalid configuration level: {config_level!r}"),
268 )
271class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder):
272 """Implements specifics required to read git style configuration files.
274 This variation behaves much like the :manpage:`git-config(1)` command, such that the
275 configuration will be read on demand based on the filepath given during
276 initialization.
278 The changes will automatically be written once the instance goes out of scope, but
279 can be triggered manually as well.
281 The configuration file will be locked if you intend to change values preventing
282 other instances to write concurrently.
284 :note:
285 The config is case-sensitive even when queried, hence section and option names
286 must match perfectly.
288 :note:
289 If used as a context manager, this will release the locked file.
290 """
292 # { Configuration
293 t_lock = LockFile
294 """The lock type determines the type of lock to use in new configuration readers.
296 They must be compatible to the :class:`~git.util.LockFile` interface.
297 A suitable alternative would be the :class:`~git.util.BlockingLockFile`.
298 """
300 re_comment = re.compile(r"^\s*[#;]")
301 # } END configuration
303 optvalueonly_source = r"\s*(?P<option>[^:=\s][^:=]*)"
305 OPTVALUEONLY = re.compile(optvalueonly_source)
307 OPTCRE = re.compile(optvalueonly_source + r"\s*(?P<vi>[:=])\s*" + r"(?P<value>.*)$")
309 del optvalueonly_source
311 _mutating_methods_ = ("add_section", "remove_section", "remove_option", "set")
312 """Names of :class:`~configparser.RawConfigParser` methods able to change the
313 instance."""
315 def __init__(
316 self,
317 file_or_files: Union[None, PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = None,
318 read_only: bool = True,
319 merge_includes: bool = True,
320 config_level: Union[Lit_config_levels, None] = None,
321 repo: Union["Repo", None] = None,
322 ) -> None:
323 """Initialize a configuration reader to read the given `file_or_files` and to
324 possibly allow changes to it by setting `read_only` False.
326 :param file_or_files:
327 A file path or file object, or a sequence of possibly more than one of them.
329 :param read_only:
330 If ``True``, the ConfigParser may only read the data, but not change it.
331 If ``False``, only a single file path or file object may be given. We will
332 write back the changes when they happen, or when the ConfigParser is
333 released. This will not happen if other configuration files have been
334 included.
336 :param merge_includes:
337 If ``True``, we will read files mentioned in ``[include]`` sections and
338 merge their contents into ours. This makes it impossible to write back an
339 individual configuration file. Thus, if you want to modify a single
340 configuration file, turn this off to leave the original dataset unaltered
341 when reading it.
343 :param repo:
344 Reference to repository to use if ``[includeIf]`` sections are found in
345 configuration files.
346 """
347 cp.RawConfigParser.__init__(self, dict_type=_OMD)
348 self._dict: Callable[..., _OMD]
349 self._defaults: _OMD
350 self._sections: _OMD
352 # Used in Python 3. Needs to stay in sync with sections for underlying
353 # implementation to work.
354 if not hasattr(self, "_proxies"):
355 self._proxies = self._dict()
357 if file_or_files is not None:
358 self._file_or_files: Union[PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = file_or_files
359 else:
360 if config_level is None:
361 if read_only:
362 self._file_or_files = [
363 get_config_path(cast(Lit_config_levels, f)) for f in CONFIG_LEVELS if f != "repository"
364 ]
365 else:
366 raise ValueError("No configuration level or configuration files specified")
367 else:
368 self._file_or_files = [get_config_path(config_level)]
370 self._read_only = read_only
371 self._dirty = False
372 self._is_initialized = False
373 self._merge_includes = merge_includes
374 self._repo = repo
375 self._lock: Union["LockFile", None] = None
376 self._acquire_lock()
378 def _acquire_lock(self) -> None:
379 if not self._read_only:
380 if not self._lock:
381 if isinstance(self._file_or_files, (str, os.PathLike)):
382 file_or_files = self._file_or_files
383 elif isinstance(self._file_or_files, (tuple, list, Sequence)):
384 raise ValueError(
385 "Write-ConfigParsers can operate on a single file only, multiple files have been passed"
386 )
387 else:
388 file_or_files = self._file_or_files.name
390 # END get filename from handle/stream
391 # Initialize lock base - we want to write.
392 self._lock = self.t_lock(file_or_files)
393 # END lock check
395 self._lock._obtain_lock()
396 # END read-only check
398 def __del__(self) -> None:
399 """Write pending changes if required and release locks."""
400 # NOTE: Only consistent in Python 2.
401 self.release()
403 def __enter__(self) -> "GitConfigParser":
404 self._acquire_lock()
405 return self
407 def __exit__(self, *args: Any) -> None:
408 self.release()
410 def release(self) -> None:
411 """Flush changes and release the configuration write lock. This instance must
412 not be used anymore afterwards.
414 In Python 3, it's required to explicitly release locks and flush changes, as
415 ``__del__`` is not called deterministically anymore.
416 """
417 # Checking for the lock here makes sure we do not raise during write()
418 # in case an invalid parser was created who could not get a lock.
419 if self.read_only or (self._lock and not self._lock._has_lock()):
420 return
422 try:
423 self.write()
424 except IOError:
425 _logger.error("Exception during destruction of GitConfigParser", exc_info=True)
426 except ReferenceError:
427 # This happens in Python 3... and usually means that some state cannot be
428 # written as the sections dict cannot be iterated. This usually happens when
429 # the interpreter is shutting down. Can it be fixed?
430 pass
431 finally:
432 if self._lock is not None:
433 self._lock._release_lock()
435 def optionxform(self, optionstr: str) -> str:
436 """Do not transform options in any way when writing."""
437 return optionstr
439 def _read(self, fp: Union[BufferedReader, IO[bytes]], fpname: str) -> None:
440 """Originally a direct copy of the Python 2.4 version of
441 :meth:`RawConfigParser._read <configparser.RawConfigParser._read>`, to ensure it
442 uses ordered dicts.
444 The ordering bug was fixed in Python 2.4, and dict itself keeps ordering since
445 Python 3.7. This has some other changes, especially that it ignores initial
446 whitespace, since git uses tabs. (Big comments are removed to be more compact.)
447 """
448 cursect = None # None, or a dictionary.
449 optname = None
450 lineno = 0
451 is_multi_line = False
452 e = None # None, or an exception.
454 def string_decode(v: str) -> str:
455 if v and v.endswith("\\"):
456 v = v[:-1]
457 # END cut trailing escapes to prevent decode error
459 return v.encode(defenc).decode("unicode_escape")
461 # END string_decode
463 while True:
464 # We assume to read binary!
465 line = fp.readline().decode(defenc)
466 if not line:
467 break
468 lineno = lineno + 1
469 # Comment or blank line?
470 if line.strip() == "" or self.re_comment.match(line):
471 continue
472 if line.split(None, 1)[0].lower() == "rem" and line[0] in "rR":
473 # No leading whitespace.
474 continue
476 # Is it a section header?
477 mo = self.SECTCRE.match(line.strip())
478 if not is_multi_line and mo:
479 sectname: str = mo.group("header").strip()
480 if sectname in self._sections:
481 cursect = self._sections[sectname]
482 elif sectname == cp.DEFAULTSECT:
483 cursect = self._defaults
484 else:
485 cursect = self._dict((("__name__", sectname),))
486 self._sections[sectname] = cursect
487 self._proxies[sectname] = None
488 # So sections can't start with a continuation line.
489 optname = None
490 # No section header in the file?
491 elif cursect is None:
492 raise cp.MissingSectionHeaderError(fpname, lineno, line)
493 # An option line?
494 elif not is_multi_line:
495 mo = self.OPTCRE.match(line)
496 if mo:
497 # We might just have handled the last line, which could contain a quotation we want to remove.
498 optname, vi, optval = mo.group("option", "vi", "value")
499 optname = self.optionxform(optname.rstrip())
501 if vi in ("=", ":") and ";" in optval and not optval.strip().startswith('"'):
502 pos = optval.find(";")
503 if pos != -1 and optval[pos - 1].isspace():
504 optval = optval[:pos]
505 optval = optval.strip()
507 if len(optval) < 2 or optval[0] != '"':
508 # Does not open quoting.
509 pass
510 elif optval[-1] != '"':
511 # Opens quoting and does not close: appears to start multi-line quoting.
512 is_multi_line = True
513 optval = string_decode(optval[1:])
514 elif optval.find("\\", 1, -1) == -1 and optval.find('"', 1, -1) == -1:
515 # Opens and closes quoting. Single line, and all we need is quote removal.
516 optval = optval[1:-1]
517 # TODO: Handle other quoted content, especially well-formed backslash escapes.
519 # Preserves multiple values for duplicate optnames.
520 cursect.add(optname, optval)
521 else:
522 # Check if it's an option with no value - it's just ignored by git.
523 if not self.OPTVALUEONLY.match(line):
524 if not e:
525 e = cp.ParsingError(fpname)
526 e.append(lineno, repr(line))
527 continue
528 else:
529 line = line.rstrip()
530 if line.endswith('"'):
531 is_multi_line = False
532 line = line[:-1]
533 # END handle quotations
534 optval = cursect.getlast(optname)
535 cursect.setlast(optname, optval + string_decode(line))
536 # END parse section or option
537 # END while reading
539 # If any parsing errors occurred, raise an exception.
540 if e:
541 raise e
543 def _has_includes(self) -> Union[bool, int]:
544 return self._merge_includes and len(self._included_paths())
546 def _included_paths(self) -> List[Tuple[str, str]]:
547 """List all paths that must be included to configuration.
549 :return:
550 The list of paths, where each path is a tuple of (option, value).
551 """
553 def _all_items(section: str) -> List[Tuple[str, str]]:
554 """Return all (key, value) pairs for a section, including duplicate keys."""
555 return [
556 (key, value)
557 for key, values in self._sections[section].items_all()
558 if key != "__name__"
559 for value in values
560 ]
562 paths = []
564 for section in self.sections():
565 if section == "include":
566 paths += _all_items(section)
568 match = CONDITIONAL_INCLUDE_REGEXP.search(section)
569 if match is None or self._repo is None:
570 continue
572 keyword = match.group(1)
573 value = match.group(2).strip()
575 if keyword in ["gitdir", "gitdir/i"]:
576 value = osp.expanduser(value)
578 if not any(value.startswith(s) for s in ["./", "/"]):
579 value = "**/" + value
580 if value.endswith("/"):
581 value += "**"
583 # Ensure that glob is always case insensitive if required.
584 if keyword.endswith("/i"):
585 value = re.sub(
586 r"[a-zA-Z]",
587 lambda m: f"[{m.group().lower()!r}{m.group().upper()!r}]",
588 value,
589 )
590 if self._repo.git_dir:
591 if fnmatch.fnmatchcase(os.fspath(self._repo.git_dir), value):
592 paths += _all_items(section)
594 elif keyword == "onbranch":
595 try:
596 branch_name = self._repo.active_branch.name
597 except TypeError:
598 # Ignore section if active branch cannot be retrieved.
599 continue
601 if fnmatch.fnmatchcase(branch_name, value):
602 paths += _all_items(section)
603 elif keyword == "hasconfig:remote.*.url":
604 for remote in self._repo.remotes:
605 if fnmatch.fnmatchcase(remote.url, value):
606 paths += _all_items(section)
607 break
608 return paths
610 def read(self) -> None: # type: ignore[override]
611 """Read the data stored in the files we have been initialized with.
613 This will ignore files that cannot be read, possibly leaving an empty
614 configuration.
616 :raise IOError:
617 If a file cannot be handled.
618 """
619 if self._is_initialized:
620 return
621 self._is_initialized = True
623 files_to_read: List[Union[PathLike, IO]] = [""]
624 if isinstance(self._file_or_files, (str, os.PathLike)):
625 # For str or Path, as str is a type of Sequence.
626 files_to_read = [self._file_or_files]
627 elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
628 # Could merge with above isinstance once runtime type known.
629 files_to_read = [self._file_or_files]
630 else: # For lists or tuples.
631 files_to_read = list(self._file_or_files)
632 # END ensure we have a copy of the paths to handle
634 seen = set(files_to_read)
635 num_read_include_files = 0
636 while files_to_read:
637 file_path = files_to_read.pop(0)
638 file_ok = False
640 if hasattr(file_path, "seek"):
641 # Must be a file-object.
642 # TODO: Replace cast with assert to narrow type, once sure.
643 file_path = cast(IO[bytes], file_path)
644 self._read(file_path, file_path.name)
645 else:
646 try:
647 with open(file_path, "rb") as fp:
648 file_ok = True
649 self._read(fp, fp.name)
650 except IOError:
651 continue
653 # Read includes and append those that we didn't handle yet. We expect all
654 # paths to be normalized and absolute (and will ensure that is the case).
655 if self._has_includes():
656 for _, include_path in self._included_paths():
657 if include_path.startswith("~"):
658 include_path = osp.expanduser(include_path)
659 if not osp.isabs(include_path):
660 if not file_ok:
661 continue
662 # END ignore relative paths if we don't know the configuration file path
663 file_path = cast(PathLike, file_path)
664 assert osp.isabs(file_path), "Need absolute paths to be sure our cycle checks will work"
665 include_path = osp.join(osp.dirname(file_path), include_path)
666 # END make include path absolute
667 include_path = osp.normpath(include_path)
668 if include_path in seen or not os.access(include_path, os.R_OK):
669 continue
670 seen.add(include_path)
671 # Insert included file to the top to be considered first.
672 files_to_read.insert(0, include_path)
673 num_read_include_files += 1
674 # END each include path in configuration file
675 # END handle includes
676 # END for each file object to read
678 # If there was no file included, we can safely write back (potentially) the
679 # configuration file without altering its meaning.
680 if num_read_include_files == 0:
681 self._merge_includes = False
683 def _write(self, fp: IO) -> None:
684 """Write an .ini-format representation of the configuration state in
685 git compatible format."""
687 def write_section(name: str, section_dict: _OMD) -> None:
688 fp.write(("[%s]\n" % name).encode(defenc))
690 values: Sequence[str] # Runtime only gets str in tests, but should be whatever _OMD stores.
691 v: str
692 for key, values in section_dict.items_all():
693 if key == "__name__":
694 continue
696 for v in values:
697 fp.write(("\t%s = %s\n" % (key, self._value_to_string(v).replace("\n", "\n\t"))).encode(defenc))
698 # END if key is not __name__
700 # END section writing
702 if self._defaults:
703 write_section(cp.DEFAULTSECT, self._defaults)
704 value: _OMD
706 for name, value in self._sections.items():
707 write_section(name, value)
709 def items(self, section_name: str) -> List[Tuple[str, str]]: # type: ignore[override]
710 """:return: list((option, value), ...) pairs of all items in the given section"""
711 return [(k, v) for k, v in super().items(section_name) if k != "__name__"]
713 def items_all(self, section_name: str) -> List[Tuple[str, List[str]]]:
714 """:return: list((option, [values...]), ...) pairs of all items in the given section"""
715 rv = _OMD(self._defaults)
717 for k, vs in self._sections[section_name].items_all():
718 if k == "__name__":
719 continue
721 if k in rv and rv.getall(k) == vs:
722 continue
724 for v in vs:
725 rv.add(k, v)
727 return rv.items_all()
729 @needs_values
730 def write(self) -> None:
731 """Write changes to our file, if there are changes at all.
733 :raise IOError:
734 If this is a read-only writer instance or if we could not obtain a file
735 lock.
736 """
737 self._assure_writable("write")
738 if not self._dirty:
739 return
741 if isinstance(self._file_or_files, (list, tuple)):
742 raise AssertionError(
743 "Cannot write back if there is not exactly a single file to write to, have %i files"
744 % len(self._file_or_files)
745 )
746 # END assert multiple files
748 if self._has_includes():
749 _logger.debug(
750 "Skipping write-back of configuration file as include files were merged in."
751 + "Set merge_includes=False to prevent this."
752 )
753 return
754 # END stop if we have include files
756 fp = self._file_or_files
758 # We have a physical file on disk, so get a lock.
759 is_file_lock = isinstance(fp, (str, os.PathLike, IOBase)) # TODO: Use PathLike (having dropped 3.5).
760 if is_file_lock and self._lock is not None: # Else raise error?
761 self._lock._obtain_lock()
763 if not hasattr(fp, "seek"):
764 fp = cast(PathLike, fp)
765 with open(fp, "wb") as fp_open:
766 self._write(fp_open)
767 else:
768 fp = cast("BytesIO", fp)
769 fp.seek(0)
770 # Make sure we do not overwrite into an existing file.
771 if hasattr(fp, "truncate"):
772 fp.truncate()
773 self._write(fp)
775 def _assure_writable(self, method_name: str) -> None:
776 if self.read_only:
777 raise IOError("Cannot execute non-constant method %s.%s" % (self, method_name))
779 def add_section(self, section: "cp._SectionName") -> None:
780 """Assures added options will stay in order."""
781 return super().add_section(section)
783 @property
784 def read_only(self) -> bool:
785 """:return: ``True`` if this instance may change the configuration file"""
786 return self._read_only
788 # FIXME: Figure out if default or return type can really include bool.
789 def get_value(
790 self,
791 section: str,
792 option: str,
793 default: Union[int, float, str, bool, None] = None,
794 ) -> Union[int, float, str, bool]:
795 """Get an option's value.
797 If multiple values are specified for this option in the section, the last one
798 specified is returned.
800 :param default:
801 If not ``None``, the given default value will be returned in case the option
802 did not exist.
804 :return:
805 A properly typed value, either int, float or string
807 :raise TypeError:
808 In case the value could not be understood.
809 Otherwise the exceptions known to the ConfigParser will be raised.
810 """
811 try:
812 valuestr = self.get(section, option)
813 except Exception:
814 if default is not None:
815 return default
816 raise
818 return self._string_to_value(valuestr)
820 def get_values(
821 self,
822 section: str,
823 option: str,
824 default: Union[int, float, str, bool, None] = None,
825 ) -> List[Union[int, float, str, bool]]:
826 """Get an option's values.
828 If multiple values are specified for this option in the section, all are
829 returned.
831 :param default:
832 If not ``None``, a list containing the given default value will be returned
833 in case the option did not exist.
835 :return:
836 A list of properly typed values, either int, float or string
838 :raise TypeError:
839 In case the value could not be understood.
840 Otherwise the exceptions known to the ConfigParser will be raised.
841 """
842 try:
843 self.sections()
844 lst = self._sections[section].getall(option)
845 except Exception:
846 if default is not None:
847 return [default]
848 raise
850 return [self._string_to_value(valuestr) for valuestr in lst]
852 def _string_to_value(self, valuestr: str) -> Union[int, float, str, bool]:
853 types = (int, float)
854 for numtype in types:
855 try:
856 val = numtype(valuestr)
857 # truncated value ?
858 if val != float(valuestr):
859 continue
860 return val
861 except (ValueError, TypeError):
862 continue
863 # END for each numeric type
865 # Try boolean values as git uses them.
866 vl = valuestr.lower()
867 if vl == "false":
868 return False
869 if vl == "true":
870 return True
872 if not isinstance(valuestr, str):
873 raise TypeError(
874 "Invalid value type: only int, long, float and str are allowed",
875 valuestr,
876 )
878 return valuestr
880 def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str:
881 if isinstance(value, (int, float, bool)):
882 return str(value)
883 return force_text(value)
885 @needs_values
886 @set_dirty_and_flush_changes
887 def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
888 """Set the given option in section to the given value.
890 This will create the section if required, and will not throw as opposed to the
891 default ConfigParser ``set`` method.
893 :param section:
894 Name of the section in which the option resides or should reside.
896 :param option:
897 Name of the options whose value to set.
899 :param value:
900 Value to set the option to. It must be a string or convertible to a string.
902 :return:
903 This instance
904 """
905 if not self.has_section(section):
906 self.add_section(section)
907 self.set(section, option, self._value_to_string(value))
908 return self
910 @needs_values
911 @set_dirty_and_flush_changes
912 def add_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
913 """Add a value for the given option in section.
915 This will create the section if required, and will not throw as opposed to the
916 default ConfigParser ``set`` method. The value becomes the new value of the
917 option as returned by :meth:`get_value`, and appends to the list of values
918 returned by :meth:`get_values`.
920 :param section:
921 Name of the section in which the option resides or should reside.
923 :param option:
924 Name of the option.
926 :param value:
927 Value to add to option. It must be a string or convertible to a string.
929 :return:
930 This instance
931 """
932 if not self.has_section(section):
933 self.add_section(section)
934 self._sections[section].add(option, self._value_to_string(value))
935 return self
937 def rename_section(self, section: str, new_name: str) -> "GitConfigParser":
938 """Rename the given section to `new_name`.
940 :raise ValueError:
941 If:
943 * `section` doesn't exist.
944 * A section with `new_name` does already exist.
946 :return:
947 This instance
948 """
949 if not self.has_section(section):
950 raise ValueError("Source section '%s' doesn't exist" % section)
951 if self.has_section(new_name):
952 raise ValueError("Destination section '%s' already exists" % new_name)
954 super().add_section(new_name)
955 new_section = self._sections[new_name]
956 for k, vs in self.items_all(section):
957 new_section.setall(k, vs)
958 # END for each value to copy
960 # This call writes back the changes, which is why we don't have the respective
961 # decorator.
962 self.remove_section(section)
963 return self