refactor(bump): cleanup related to update_version_file

Author: bearomorphism

commitizen/bump.py

@@ -3,11 +3,13 @@
 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 TextIO, cast
+
+from typing_extensions import Never
 
 from commitizen.defaults import BUMP_MESSAGE, ENCODING, MAJOR, MINOR, PATCH
 from commitizen.exceptions import CurrentVersionNotFoundError
@@ -75,36 +77,45 @@ def update_version_in_files(
 
     Returns the list of updated 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:
+
+    for path, pattern in _resolve_files_and_regexes(files, current_version):
+
+        def raise_not_found_error() -> 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."
             )
 
+        def bump_line(line: str) -> str:
+            return (
+                line.replace(current_version, new_version)
+                if pattern.search(line)
+                else line
+            )
+
+        with open(path, encoding=encoding) as file:
+            bumped_version_file_content = _get_bumped_version_file_content(
+                file,
+                bump_line,
+                raise_not_found_error if check_consistency else None,
+            )
+
         # 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 +124,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 _get_bumped_version_file_content(
+    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(