# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
#
# This module is part of GitPython and is released under the
# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
"""Parser for reading and writing configuration files."""
import abc
import configparser as cp
import fnmatch
from functools import wraps
import inspect
from io import BufferedReader, IOBase
import logging
import os
import os.path as osp
import re
import sys
from git.compat import defenc, force_text
from git.util import LockFile
# typing-------------------------------------------------------
from typing import (
Any,
Callable,
Generic,
IO,
List,
Dict,
Sequence,
TYPE_CHECKING,
Tuple,
TypeVar,
Union,
cast,
)
from git.types import Lit_config_levels, ConfigLevels_Tup, PathLike, assert_never, _T
if TYPE_CHECKING:
from git.repo.base import Repo
from io import BytesIO
T_ConfigParser = TypeVar("T_ConfigParser", bound="GitConfigParser")
T_OMD_value = TypeVar("T_OMD_value", str, bytes, int, float, bool)
if sys.version_info[:3] < (3, 7, 2):
# typing.Ordereddict not added until Python 3.7.2.
from collections import OrderedDict
OrderedDict_OMD = OrderedDict
else:
from typing import OrderedDict
OrderedDict_OMD = OrderedDict[str, List[T_OMD_value]] # type: ignore[assignment, misc]
# -------------------------------------------------------------
__all__ = ("GitConfigParser", "SectionConstraint")
log = logging.getLogger("git.config")
log.addHandler(logging.NullHandler())
# The configuration level of a configuration file.
CONFIG_LEVELS: ConfigLevels_Tup = ("system", "user", "global", "repository")
# Section pattern to detect conditional includes.
# https://git-scm.com/docs/git-config#_conditional_includes
CONDITIONAL_INCLUDE_REGEXP = re.compile(r"(? 1 and optval[0] == '"' and optval[-1] != '"':
is_multi_line = True
optval = string_decode(optval[1:])
# END handle multi-line
# Preserves multiple values for duplicate optnames.
cursect.add(optname, optval)
else:
# Check if it's an option with no value - it's just ignored by git.
if not self.OPTVALUEONLY.match(line):
if not e:
e = cp.ParsingError(fpname)
e.append(lineno, repr(line))
continue
else:
line = line.rstrip()
if line.endswith('"'):
is_multi_line = False
line = line[:-1]
# END handle quotations
optval = cursect.getlast(optname)
cursect.setlast(optname, optval + string_decode(line))
# END parse section or option
# END while reading
# If any parsing errors occurred, raise an exception.
if e:
raise e
def _has_includes(self) -> Union[bool, int]:
return self._merge_includes and len(self._included_paths())
def _included_paths(self) -> List[Tuple[str, str]]:
"""List all paths that must be included to configuration.
:return: The list of paths, where each path is a tuple of ``(option, value)``.
"""
paths = []
for section in self.sections():
if section == "include":
paths += self.items(section)
match = CONDITIONAL_INCLUDE_REGEXP.search(section)
if match is None or self._repo is None:
continue
keyword = match.group(1)
value = match.group(2).strip()
if keyword in ["gitdir", "gitdir/i"]:
value = osp.expanduser(value)
if not any(value.startswith(s) for s in ["./", "/"]):
value = "**/" + value
if value.endswith("/"):
value += "**"
# Ensure that glob is always case insensitive if required.
if keyword.endswith("/i"):
value = re.sub(
r"[a-zA-Z]",
lambda m: "[{}{}]".format(m.group().lower(), m.group().upper()),
value,
)
if self._repo.git_dir:
if fnmatch.fnmatchcase(str(self._repo.git_dir), value):
paths += self.items(section)
elif keyword == "onbranch":
try:
branch_name = self._repo.active_branch.name
except TypeError:
# Ignore section if active branch cannot be retrieved.
continue
if fnmatch.fnmatchcase(branch_name, value):
paths += self.items(section)
return paths
def read(self) -> None: # type: ignore[override]
"""Read the data stored in the files we have been initialized with.
This will ignore files that cannot be read, possibly leaving an empty
configuration.
:return: Nothing
:raise IOError: If a file cannot be handled
"""
if self._is_initialized:
return
self._is_initialized = True
files_to_read: List[Union[PathLike, IO]] = [""]
if isinstance(self._file_or_files, (str, os.PathLike)):
# For str or Path, as str is a type of Sequence.
files_to_read = [self._file_or_files]
elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
# Could merge with above isinstance once runtime type known.
files_to_read = [self._file_or_files]
else: # for lists or tuples
files_to_read = list(self._file_or_files)
# END ensure we have a copy of the paths to handle
seen = set(files_to_read)
num_read_include_files = 0
while files_to_read:
file_path = files_to_read.pop(0)
file_ok = False
if hasattr(file_path, "seek"):
# Must be a file-object.
file_path = cast(IO[bytes], file_path) # TODO: Replace with assert to narrow type, once sure.
self._read(file_path, file_path.name)
else:
# Assume a path if it is not a file-object.
file_path = cast(PathLike, file_path)
try:
with open(file_path, "rb") as fp:
file_ok = True
self._read(fp, fp.name)
except IOError:
continue
# Read includes and append those that we didn't handle yet.
# We expect all paths to be normalized and absolute (and will ensure that is the case).
if self._has_includes():
for _, include_path in self._included_paths():
if include_path.startswith("~"):
include_path = osp.expanduser(include_path)
if not osp.isabs(include_path):
if not file_ok:
continue
# END ignore relative paths if we don't know the configuration file path
file_path = cast(PathLike, file_path)
assert osp.isabs(file_path), "Need absolute paths to be sure our cycle checks will work"
include_path = osp.join(osp.dirname(file_path), include_path)
# END make include path absolute
include_path = osp.normpath(include_path)
if include_path in seen or not os.access(include_path, os.R_OK):
continue
seen.add(include_path)
# Insert included file to the top to be considered first.
files_to_read.insert(0, include_path)
num_read_include_files += 1
# END each include path in configuration file
# END handle includes
# END for each file object to read
# If there was no file included, we can safely write back (potentially) the
# configuration file without altering its meaning.
if num_read_include_files == 0:
self._merge_includes = False
def _write(self, fp: IO) -> None:
"""Write an .ini-format representation of the configuration state in
git compatible format."""
def write_section(name: str, section_dict: _OMD) -> None:
fp.write(("[%s]\n" % name).encode(defenc))
values: Sequence[str] # Runtime only gets str in tests, but should be whatever _OMD stores.
v: str
for key, values in section_dict.items_all():
if key == "__name__":
continue
for v in values:
fp.write(("\t%s = %s\n" % (key, self._value_to_string(v).replace("\n", "\n\t"))).encode(defenc))
# END if key is not __name__
# END section writing
if self._defaults:
write_section(cp.DEFAULTSECT, self._defaults)
value: _OMD
for name, value in self._sections.items():
write_section(name, value)
def items(self, section_name: str) -> List[Tuple[str, str]]: # type: ignore[override]
""":return: list((option, value), ...) pairs of all items in the given section"""
return [(k, v) for k, v in super().items(section_name) if k != "__name__"]
def items_all(self, section_name: str) -> List[Tuple[str, List[str]]]:
""":return: list((option, [values...]), ...) pairs of all items in the given section"""
rv = _OMD(self._defaults)
for k, vs in self._sections[section_name].items_all():
if k == "__name__":
continue
if k in rv and rv.getall(k) == vs:
continue
for v in vs:
rv.add(k, v)
return rv.items_all()
@needs_values
def write(self) -> None:
"""Write changes to our file, if there are changes at all.
:raise IOError: If this is a read-only writer instance or if we could not obtain
a file lock"""
self._assure_writable("write")
if not self._dirty:
return
if isinstance(self._file_or_files, (list, tuple)):
raise AssertionError(
"Cannot write back if there is not exactly a single file to write to, have %i files"
% len(self._file_or_files)
)
# END assert multiple files
if self._has_includes():
log.debug(
"Skipping write-back of configuration file as include files were merged in."
+ "Set merge_includes=False to prevent this."
)
return
# END stop if we have include files
fp = self._file_or_files
# We have a physical file on disk, so get a lock.
is_file_lock = isinstance(fp, (str, os.PathLike, IOBase)) # TODO: Use PathLike (having dropped 3.5).
if is_file_lock and self._lock is not None: # Else raise error?
self._lock._obtain_lock()
if not hasattr(fp, "seek"):
fp = cast(PathLike, fp)
with open(fp, "wb") as fp_open:
self._write(fp_open)
else:
fp = cast("BytesIO", fp)
fp.seek(0)
# Make sure we do not overwrite into an existing file.
if hasattr(fp, "truncate"):
fp.truncate()
self._write(fp)
def _assure_writable(self, method_name: str) -> None:
if self.read_only:
raise IOError("Cannot execute non-constant method %s.%s" % (self, method_name))
def add_section(self, section: str) -> None:
"""Assures added options will stay in order"""
return super().add_section(section)
@property
def read_only(self) -> bool:
""":return: True if this instance may change the configuration file"""
return self._read_only
# FIXME: Figure out if default or return type can really include bool.
def get_value(
self,
section: str,
option: str,
default: Union[int, float, str, bool, None] = None,
) -> Union[int, float, str, bool]:
"""Get an option's value.
If multiple values are specified for this option in the section, the
last one specified is returned.
:param default:
If not None, the given default value will be returned in case
the option did not exist
:return: a properly typed value, either int, float or string
:raise TypeError: in case the value could not be understood
Otherwise the exceptions known to the ConfigParser will be raised.
"""
try:
valuestr = self.get(section, option)
except Exception:
if default is not None:
return default
raise
return self._string_to_value(valuestr)
def get_values(
self,
section: str,
option: str,
default: Union[int, float, str, bool, None] = None,
) -> List[Union[int, float, str, bool]]:
"""Get an option's values.
If multiple values are specified for this option in the section, all are
returned.
:param default:
If not None, a list containing the given default value will be
returned in case the option did not exist
:return: a list of properly typed values, either int, float or string
:raise TypeError: in case the value could not be understood
Otherwise the exceptions known to the ConfigParser will be raised.
"""
try:
self.sections()
lst = self._sections[section].getall(option)
except Exception:
if default is not None:
return [default]
raise
return [self._string_to_value(valuestr) for valuestr in lst]
def _string_to_value(self, valuestr: str) -> Union[int, float, str, bool]:
types = (int, float)
for numtype in types:
try:
val = numtype(valuestr)
# truncated value ?
if val != float(valuestr):
continue
return val
except (ValueError, TypeError):
continue
# END for each numeric type
# Try boolean values as git uses them.
vl = valuestr.lower()
if vl == "false":
return False
if vl == "true":
return True
if not isinstance(valuestr, str):
raise TypeError(
"Invalid value type: only int, long, float and str are allowed",
valuestr,
)
return valuestr
def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str:
if isinstance(value, (int, float, bool)):
return str(value)
return force_text(value)
@needs_values
@set_dirty_and_flush_changes
def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
"""Set the given option in section to the given value.
This will create the section if required, and will not throw as opposed to the
default ConfigParser 'set' method.
:param section: Name of the section in which the option resides or should reside
:param option: Name of the options whose value to set
:param value: Value to set the option to. It must be a string or convertible to
a string.
:return: This instance
"""
if not self.has_section(section):
self.add_section(section)
self.set(section, option, self._value_to_string(value))
return self
@needs_values
@set_dirty_and_flush_changes
def add_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
"""Add a value for the given option in section.
This will create the section if required, and will not throw as opposed to the default
ConfigParser 'set' method. The value becomes the new value of the option as returned
by 'get_value', and appends to the list of values returned by 'get_values`'.
:param section: Name of the section in which the option resides or should reside
:param option: Name of the option
:param value: Value to add to option. It must be a string or convertible
to a string
:return: This instance
"""
if not self.has_section(section):
self.add_section(section)
self._sections[section].add(option, self._value_to_string(value))
return self
def rename_section(self, section: str, new_name: str) -> "GitConfigParser":
"""Rename the given section to new_name.
:raise ValueError: If ``section`` doesn't exist
:raise ValueError: If a section with ``new_name`` does already exist
:return: This instance
"""
if not self.has_section(section):
raise ValueError("Source section '%s' doesn't exist" % section)
if self.has_section(new_name):
raise ValueError("Destination section '%s' already exists" % new_name)
super().add_section(new_name)
new_section = self._sections[new_name]
for k, vs in self.items_all(section):
new_section.setall(k, vs)
# END for each value to copy
# This call writes back the changes, which is why we don't have the respective decorator.
self.remove_section(section)
return self