"""Utility class wrapping base-class methods into decorators that assure read-only

def __new__(cls, name: str, bases: Tuple, clsdict: Dict[str, Any]) -> "MetaParserBuilder":

"""Equip all base-class methods with a needs_values decorator, and all non-const

kmm = "_mutating_methods_"

if kmm in clsdict:

mutating_methods = clsdict[kmm]

for base in bases:

methods = (t for t in inspect.getmembers(base, inspect.isroutine) if not t[0].startswith("_"))

for method_name, method in methods:

if method_name in clsdict:

continue

method_with_values = needs_values(method)

if method_name in mutating_methods:

method_with_values = set_dirty_and_flush_changes(method_with_values)

# END mutating methods handling

clsdict[method_name] = method_with_values

# END for each name/method pair

# END for each base

# END if mutating methods configuration is set

new_type = super().__new__(cls, name, bases, clsdict)

"""Return a method for ensuring we read values (on demand) before we try to access

@wraps(func)

def assure_data_present(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:

self.read()

return func(self, *args, **kwargs)

# END wrapper method

"""Return a method that checks whether given non constant function may be called.

def flush_changes(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T:

rval = non_const_func(self, *args, **kwargs)

self._dirty = True

self.write()

return rval

# END wrapper method

flush_changes.__name__ = non_const_func.__name__

"""Constrains a ConfigParser to only option commands which are constrained to

__slots__ = ("_config", "_section_name")

_valid_attrs_ = (

"get_value",

"set_value",

"get",

"set",

"getint",

"getfloat",

"getboolean",

"has_option",

"remove_section",

"remove_option",

"options",

)

def __init__(self, config: T_ConfigParser, section: str) -> None:

self._config = config

self._section_name = section

def __del__(self) -> None:

# Yes, for some reason, we have to call it explicitly for it to work in PY3 !

# Apparently __del__ doesn't get call anymore if refcount becomes 0

# Ridiculous ... .

self._config.release()

def __getattr__(self, attr: str) -> Any:

if attr in self._valid_attrs_:

return lambda *args, **kwargs: self._call_config(attr, *args, **kwargs)

return super().__getattribute__(attr)

def _call_config(self, method: str, *args: Any, **kwargs: Any) -> Any:

"""Call the configuration at the given method which must take a section name as

return getattr(self._config, method)(self._section_name, *args, **kwargs)

@property

def config(self) -> T_ConfigParser:

"""return: ConfigParser instance we constrain"""

return self._config

def release(self) -> None:

"""Equivalent to :meth:`GitConfigParser.release`, which is called on our

return self._config.release()

def __enter__(self) -> "SectionConstraint[T_ConfigParser]":

self._config.__enter__()

return self

def __exit__(self, exception_type: str, exception_value: str, traceback: str) -> None:

"""Ordered multi-dict matching config names while retaining their first spelling."""

def __init__(self, *args: Any, **kwargs: Any) -> None:

self._keymap: Dict[str, str] = {}

super().__init__(*args, **kwargs)

def _key(self, key: str) -> str:

stored = self._keymap.get(_normalize_name(key), key)

return stored if super().__contains__(stored) else key

def __contains__(self, key: object) -> bool:

return isinstance(key, str) and super().__contains__(self._key(key))

def __delitem__(self, key: str) -> None:

super().__delitem__(self._key(key))

del self._keymap[_normalize_name(key)]

def __setitem__(self, key: str, value: _T) -> None:

self.setall(key, [value])

def clear(self) -> None:

super().clear()

self._keymap.clear()

def add(self, key: str, value: Any) -> None:

if key not in self:

self[key] = value

return

self.getall(key).append(value)

def setall(self, key: str, values: List[_T]) -> None:

key = self._key(key)

super().__setitem__(key, values)

self._keymap[_normalize_name(key)] = key

def __getitem__(self, key: str) -> Any:

return super().__getitem__(self._key(key))[-1]

def getlast(self, key: str) -> Any:

return self[key]

def setlast(self, key: str, value: Any) -> None:

if key not in self:

self[key] = value

return

self.getall(key)[-1] = value

def get(self, key: str, default: Union[_T, None] = None) -> Union[_T, None]: # type: ignore[override]

return super().get(self._key(key), [default])[-1]

def getall(self, key: str) -> List[_T]:

return super().__getitem__(self._key(key))

def items(self) -> List[Tuple[str, _T]]: # type: ignore[override]

"""List of (key, last value for key)."""

return [(k, self[k]) for k in self]

def items_all(self) -> List[Tuple[str, List[_T]]]:

"""List of (key, list of values for key)."""

# We do not support an absolute path of the gitconfig on Windows.

# Use the global config instead.

if sys.platform == "win32" and config_level == "system":

config_level = "global"

if config_level == "system":

return "/etc/gitconfig"

elif config_level == "user":

config_home = os.environ.get("XDG_CONFIG_HOME") or osp.join(os.environ.get("HOME", "~"), ".config")

return osp.normpath(osp.expanduser(osp.join(config_home, "git", "config")))

elif config_level == "global":

return osp.normpath(osp.expanduser("~/.gitconfig"))

elif config_level == "repository":

raise ValueError("No repo to get repository configuration from. Use Repo._get_config_path")

else:

# Should not reach here. Will raise ValueError if does. Static typing will warn

# about missing elifs.

assert_never( # type: ignore[unreachable]

config_level,

ValueError(f"Invalid configuration level: {config_level!r}"),

"""Implements specifics required to read git style configuration files.

# { Configuration

t_lock = LockFile

"""The lock type determines the type of lock to use in new configuration readers.

re_comment = re.compile(r"^\s*[#;]")

# } END configuration

optvalueonly_source = r"\s*(?P<option>[^:=\s#;][^:=#;]*)"

OPTVALUEONLY = re.compile(optvalueonly_source)

# The option name class [^:=#;]* already consumes any spaces up to the ":" or "=",

# so a second \s* before the indicator would overlap it and backtrack quadratically

# on a line that never reaches an indicator (for example a key followed by a long

# whitespace run). Drop the redundant \s*; the name is right-stripped after parsing.

OPTCRE = re.compile(optvalueonly_source + r"(?P<vi>[:=])\s*" + r"(?P<value>.*)$")

del optvalueonly_source

_mutating_methods_ = ("add_section", "remove_section", "remove_option", "set")

"""Names of :class:`~configparser.RawConfigParser` methods able to change the

def __init__(

self,

file_or_files: Union[None, PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = None,

read_only: bool = True,

merge_includes: bool = True,

config_level: Union[Lit_config_levels, None] = None,

repo: Union["Repo", None] = None,

) -> None:

"""Initialize a configuration reader to read the given `file_or_files` and to

cp.RawConfigParser.__init__(self, dict_type=cast(Any, _OMD), allow_no_value=True)

self._dict: Callable[..., _OMD]

self._defaults: _OMD

self._sections: _OMD

# Used in Python 3. Needs to stay in sync with sections for underlying

# implementation to work.

if not hasattr(self, "_proxies"):

self._proxies = self._dict()

if file_or_files is not None:

self._file_or_files: Union[PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = file_or_files

else:

if config_level is None:

if read_only:

self._file_or_files = [

get_config_path(cast(Lit_config_levels, f)) for f in CONFIG_LEVELS if f != "repository"

]

else:

raise ValueError("No configuration level or configuration files specified")

else:

self._file_or_files = [get_config_path(config_level)]

self._read_only = read_only

self._dirty = False

self._is_initialized = False

self._merge_includes = merge_includes

self._repo = repo

self._lock: Union["LockFile", None] = None

self._acquire_lock()

def _acquire_lock(self) -> None:

if not self._read_only:

if not self._lock:

if isinstance(self._file_or_files, (str, os.PathLike)):

file_or_files = self._file_or_files

elif isinstance(self._file_or_files, (tuple, list, Sequence)):

raise ValueError(

"Write-ConfigParsers can operate on a single file only, multiple files have been passed"

)

else:

file_or_files = self._file_or_files.name

# END get filename from handle/stream

# Initialize lock base - we want to write.

self._lock = self.t_lock(file_or_files)

# END lock check

self._lock._obtain_lock()

# END read-only check

def __del__(self) -> None:

"""Write pending changes if required and release locks."""

# NOTE: Only consistent in Python 2.

self.release()

def __enter__(self) -> "GitConfigParser":

self._acquire_lock()

return self

def __exit__(self, *args: Any) -> None:

self.release()

def release(self) -> None:

"""Flush changes and release the configuration write lock. This instance must

# Checking for the lock here makes sure we do not raise during write()

# in case an invalid parser was created who could not get a lock.

if self.read_only or (self._lock and not self._lock._has_lock()):

return

try:

self.write()

except OSError:

_logger.error("Exception during destruction of GitConfigParser", exc_info=True)

except ReferenceError:

# This happens in Python 3... and usually means that some state cannot be

# written as the sections dict cannot be iterated. This usually happens when

# the interpreter is shutting down. Can it be fixed?

pass

finally:

if self._lock is not None:

self._lock._release_lock()

def optionxform(self, optionstr: str) -> str:

"""Do not transform options in any way when writing."""

return optionstr

def _read(self, fp: Union[BufferedReader, IO[bytes]], fpname: str) -> None:

"""Originally a direct copy of the Python 2.4 version of

cursect = None # None, or a dictionary.

optname = None

lineno = 0

is_multi_line = False

e = None # None, or an exception.

def string_decode(v: str) -> str:

if v and v.endswith("\\"):

v = v[:-1]

# END cut trailing escapes to prevent decode error

escapes = {"b": "\b", "n": "\n", "r": "\r", "t": "\t", '"': '"', "\\": "\\"}

return re.sub(r"\\(.)", lambda match: escapes.get(match.group(1), match.group(0)), v)

# END string_decode

def is_line_continuation(value: str) -> bool:

quoted = escaped = False

for char in value:

if escaped:

escaped = False

elif char == "\\":

escaped = True

elif char == '"':

quoted = not quoted

elif char in "#;" and not quoted:

return False

return escaped

def strip_inline_comment(value: str) -> Tuple[str, bool]:

"""Cut an unquoted ``#`` or ``;`` comment and report whether a quote is open.

quoted = escaped = False

for index, char in enumerate(value):

if escaped:

escaped = False

elif char == "\\":

escaped = True

elif char == '"':

quoted = not quoted

elif char in "#;" and not quoted:

return value[:index], False

return value, quoted

def parse_value(value: str) -> str:

parsed: List[str] = []

whitespace: List[str] = []

quoted = escaped = False

escapes = {"b": "\b", "n": "\n", "t": "\t", '"': '"', "\\": "\\"}

for char in value:

if escaped:

parsed.append(escapes.get(char, "\\" + char))

escaped = False

continue

if char.isspace() and not quoted:

if parsed:

whitespace.append(char)

continue

if char in "#;" and not quoted:

break

parsed.extend(whitespace)

whitespace.clear()

if char == "\\":

escaped = True

elif char == '"':

quoted = not quoted

else:

parsed.append(char)

return "".join(parsed)

while True:

# We assume to read binary!

raw_line = fp.readline()

if not raw_line:

break

if lineno == 0 and raw_line.startswith(b"\xef\xbb\xbf"):

# A UTF-8 BOM is not part of the content. git skips it, so a

# config file written by a Windows editor still parses.

raw_line = raw_line[3:]

line = raw_line.decode(defenc)

lineno = lineno + 1

# Comment or blank line?

if line.strip() == "" or self.re_comment.match(line):

continue

if line.split(None, 1)[0].lower() == "rem" and line[0] in "rR":

# No leading whitespace.

continue

# Is it a section header?

mo = self.SECTCRE.match(line.strip())

if not is_multi_line and mo:

sectname: str = mo.group("header").strip()

if sectname in self._sections:

cursect = self._sections[sectname]

elif sectname == cp.DEFAULTSECT:

cursect = self._defaults

else:

cursect = self._dict((("__name__", sectname),))

self._sections[sectname] = cursect

self._proxies[sectname] = None

# So sections can't start with a continuation line.

optname = None

# No section header in the file?

elif cursect is None:

raise cp.MissingSectionHeaderError(fpname, lineno, line)

# An option line?

elif not is_multi_line:

mo = self.OPTCRE.match(line)

if mo:

# We might just have handled the last line, which could contain a quotation we want to remove.

optname, vi, optval = mo.group("option", "vi", "value")

optname = self.optionxform(optname.rstrip())

optval, quote_open = strip_inline_comment(optval)

optval = optval.strip()

if len(optval) < 2 or optval[0] != '"':

# Does not open quoting.

# A value ending in an odd number of backslashes

# continues on the next line, exactly as git does: the

# final backslash and the newline are removed and the

# next line is appended before the complete value is

# parsed. An even number means the last backslash is

# escaped and the value ends there.

continued = False

while True:

if not is_line_continuation(optval):

break

continuation = fp.readline()

if not continuation:

# Backslash at end of file: git drops it.

optval = optval[:-1]

break

lineno = lineno + 1

joined = continuation.decode(defenc)

while joined.endswith("\n") or joined.endswith("\r"):

joined = joined[:-1]

optval = optval[:-1] + joined

continued = True

if continued:

optval = parse_value(optval)

elif quote_open:

# Opens quoting and does not close: appears to start multi-line quoting.

is_multi_line = True

optval = string_decode(optval[1:])

elif re.search(r'(?:^|[^\\])(?:\\\\)*"', optval[1:-1]):

# Preserve values containing additional unescaped quotes.

pass

else:

# Opens and closes quoting.

optval = string_decode(optval[1:-1])

# Preserves multiple values for duplicate optnames.

cursect.add(optname, optval)

else:

# A valueless option is an implicit boolean true, not an empty value.

mo = self.OPTVALUEONLY.fullmatch(line)

if mo:

optname = self.optionxform(mo.group("option").rstrip())

cursect.add(optname, None)

else:

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.

def _all_items(section: str) -> List[Tuple[str, str]]:

"""Return all (key, value) pairs for a section, including duplicate keys."""

return [

(key, value)

for key, values in self._sections[section].items_all()

if key != "__name__"

for value in values

if value is not None

]

paths = []

for section in self.sections():

normalized_section = _normalize_name(section)

if normalized_section == "include":

paths += _all_items(section)

match = CONDITIONAL_INCLUDE_REGEXP.search(normalized_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)

git_dir = os.fspath(self._repo.git_dir) if self._repo.git_dir else None

if sys.platform == "win32":

git_dir = git_dir.replace("\\", "/") if git_dir else None

drive, _tail = osp.splitdrive(value)

if not drive and 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: f"[{m.group().lower()!r}{m.group().upper()!r}]",

value,

)

if git_dir and fnmatch.fnmatchcase(git_dir, value):

paths += _all_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 += _all_items(section)

elif keyword == "hasconfig:remote.*.url":

for remote in self._repo.remotes:

if fnmatch.fnmatchcase(remote.url, value):

paths += _all_items(section)

break

return paths

def read(self) -> None: # type: ignore[override]

"""Read the data stored in the files we have been initialized with.

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

files_to_read = [osp.abspath(path) if isinstance(path, (str, os.PathLike)) else path for path in files_to_read]

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.

# TODO: Replace cast with assert to narrow type, once sure.

file_path = cast(IO[bytes], file_path)

self._read(file_path, file_path.name)

else:

try:

with open(file_path, "rb") as fp:

file_ok = True

self._read(fp, fp.name)

except OSError:

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

def write_section(name: str, section_dict: _OMD) -> None:

fp.write(("[%s]\n" % name).encode(defenc))

values: List[Any]

v: Any

for key, values in section_dict.items_all():

if key == "__name__":

continue

for v in values:

if v is None:

fp.write(("\t%s\n" % key).encode(defenc))

continue

value = self._value_to_string(v)

if any(char in value for char in '\n\t\b\\"#;') or value[:1].isspace() or value[-1:].isspace():

value = value.replace("\\", "\\\\").replace('"', '\\"')

value = '"%s\\\n"' % value.replace("\n", "\\n").replace("\t", "\\t").replace("\b", "\\b")

fp.write(("\t%s = %s\n" % (key, value)).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, Union[str, None]]]: # 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[Union[str, None]]]]:

""":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.

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():

_logger.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

sections: List[_OMD] = [self._defaults]

section: _OMD

stored_section: _OMD

values: List[Any]

raw_value: Any

for _, stored_section in self._sections.items():

sections.append(stored_section)

for section in sections:

for key, values in section.items_all():

if key != "__name__":

for raw_value in values:

if raw_value is None:

continue

if "\r" in self._value_to_string(raw_value) or "\x00" in self._value_to_string(raw_value):

raise ValueError("Git config values must not contain CR or NUL")

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 OSError(f"Cannot execute non-constant method {self}.{method_name}")

def add_section(self, section: "cp._SectionName") -> None:

"""Assures added options will stay in order."""

self._assure_config_name_safe(section, "section")

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

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.

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.

try:

self.sections()

lst = self._sections[section].getall(option)

except Exception:

if default is not None:

return [default]

raise