refactor(bump): cleanup related to update_version_file

Author: bearomorphism

commitizen/bump.py

@@ -3,11 +3,11 @@
 import os
 import re
 from collections import OrderedDict
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from glob import iglob
 from logging import getLogger
 from string import Template
-from typing import cast
+from typing import Never, TextIO, cast
 
 from commitizen.defaults import BUMP_MESSAGE, ENCODING, MAJOR, MINOR, PATCH
 from commitizen.exceptions import CurrentVersionNotFoundError
@@ -77,34 +77,43 @@ def update_version_in_files(
     """
     # TODO: separate check step and write step
     updated = []
-    for path, regex in _files_and_regexes(files, current_version):
-        current_version_found, version_file = _bump_with_regex(
-            path,
-            current_version,
-            new_version,
-            regex,
-            encoding=encoding,
-        )
-
-        if check_consistency and not current_version_found:
-            raise CurrentVersionNotFoundError(
-                f"Current version {current_version} is not found in {path}.\n"
-                "The version defined in commitizen configuration and the ones in "
-                "version_files are possibly inconsistent."
+
+    for path, pattern in _resolve_files_and_regexes(files, current_version):
+        error_on_not_found = None
+        if check_consistency:
+
+            def fn() -> Never:
+                raise CurrentVersionNotFoundError(
+                    f"Current version {current_version} is not found in {path}.\n"
+                    "The version defined in commitizen configuration and the ones in "
+                    "version_files are possibly inconsistent."
+                )
+
+            error_on_not_found = fn
+
+        with open(path, encoding=encoding) as file:
+            bumped_version_file_content = _bump_line_by_line(
+                file,
+                lambda line: line.replace(current_version, new_version)
+                if pattern.search(line)
+                else line,
+                error_on_not_found,
             )
 
         # Write the file out again
         with smart_open(path, "w", encoding=encoding) as file:
-            file.write(version_file)
+            file.write(bumped_version_file_content)
         updated.append(path)
     return updated
 
 
-def _files_and_regexes(patterns: Iterable[str], version: str) -> list[tuple[str, str]]:
+def _resolve_files_and_regexes(
+    patterns: Iterable[str], version: str
+) -> list[tuple[str, re.Pattern]]:
     """
     Resolve all distinct files with their regexp from a list of glob patterns with optional regexp
     """
-    out: set[tuple[str, str]] = set()
+    out: set[tuple[str, re.Pattern]] = set()
     for pattern in patterns:
         drive, tail = os.path.splitdrive(pattern)
         path, _, regex = tail.partition(":")
@@ -113,33 +122,30 @@ def _files_and_regexes(patterns: Iterable[str], version: str) -> list[tuple[str,
             regex = re.escape(version)
 
         for file in iglob(filepath):
-            out.add((file, regex))
+            out.add((file, re.compile(regex)))
 
     return sorted(out)
 
 
-def _bump_with_regex(
-    version_filepath: str,
-    current_version: str,
-    new_version: str,
-    regex: str,
-    encoding: str = ENCODING,
-) -> tuple[bool, str]:
+def _bump_line_by_line(
+    version_file: TextIO,
+    bump_line: Callable[[str], str],
+    error_on_not_found: Callable[[], Never] | None,
+) -> str:
     current_version_found = False
     lines = []
-    pattern = re.compile(regex)
-    with open(version_filepath, encoding=encoding) as f:
-        for line in f:
-            if not pattern.search(line):
-                lines.append(line)
-                continue
-
-            bumped_line = line.replace(current_version, new_version)
-            if bumped_line != line:
-                current_version_found = True
-            lines.append(bumped_line)
-
-    return current_version_found, "".join(lines)
+
+    for line in version_file:
+        bumped_line = bump_line(line)
+        if bumped_line != line:
+            current_version_found = True
+
+        lines.append(bumped_line)
+
+    if error_on_not_found is not None and not current_version_found:
+        error_on_not_found()
+
+    return "".join(lines)
 
 
 def create_commit_message(