3
3
import os
4
4
import re
5
5
from collections import OrderedDict
6
- from collections .abc import Iterable
6
+ from collections .abc import Callable , Iterable
7
7
from glob import iglob
8
8
from logging import getLogger
9
9
from string import Template
10
- from typing import cast
10
+ from typing import Never , TextIO , cast
11
11
12
12
from commitizen .defaults import BUMP_MESSAGE , ENCODING , MAJOR , MINOR , PATCH
13
13
from commitizen .exceptions import CurrentVersionNotFoundError
@@ -77,34 +77,45 @@ def update_version_in_files(
77
77
"""
78
78
# TODO: separate check step and write step
79
79
updated = []
80
- for path , regex in _files_and_regexes (files , current_version ):
81
- current_version_found , version_file = _bump_with_regex (
82
- path ,
83
- current_version ,
84
- new_version ,
85
- regex ,
86
- encoding = encoding ,
80
+
81
+ for path , pattern in _resolve_files_and_regexes (files , current_version ):
82
+ error = CurrentVersionNotFoundError (
83
+ f"Current version { current_version } is not found in { path } .\n "
84
+ "The version defined in commitizen configuration and the ones in "
85
+ "version_files are possibly inconsistent."
86
+ )
87
+ error_on_not_found = (
88
+ None if not check_consistency else lambda : (_ for _ in ()).throw (error )
87
89
)
88
90
89
- if check_consistency and not current_version_found :
90
- raise CurrentVersionNotFoundError (
91
- f"Current version { current_version } is not found in { path } .\n "
92
- "The version defined in commitizen configuration and the ones in "
93
- "version_files are possibly inconsistent."
91
+ def bump_line (line : str ) -> str :
92
+ return (
93
+ line .replace (current_version , new_version )
94
+ if pattern .search (line )
95
+ else line
96
+ )
97
+
98
+ with open (path , encoding = encoding ) as file :
99
+ bumped_version_file_content = _get_bumped_version_file_content (
100
+ file ,
101
+ bump_line ,
102
+ error_on_not_found ,
94
103
)
95
104
96
105
# Write the file out again
97
106
with smart_open (path , "w" , encoding = encoding ) as file :
98
- file .write (version_file )
107
+ file .write (bumped_version_file_content )
99
108
updated .append (path )
100
109
return updated
101
110
102
111
103
- def _files_and_regexes (patterns : Iterable [str ], version : str ) -> list [tuple [str , str ]]:
112
+ def _resolve_files_and_regexes (
113
+ patterns : Iterable [str ], version : str
114
+ ) -> list [tuple [str , re .Pattern ]]:
104
115
"""
105
116
Resolve all distinct files with their regexp from a list of glob patterns with optional regexp
106
117
"""
107
- out : set [tuple [str , str ]] = set ()
118
+ out : set [tuple [str , re . Pattern ]] = set ()
108
119
for pattern in patterns :
109
120
drive , tail = os .path .splitdrive (pattern )
110
121
path , _ , regex = tail .partition (":" )
@@ -113,33 +124,30 @@ def _files_and_regexes(patterns: Iterable[str], version: str) -> list[tuple[str,
113
124
regex = re .escape (version )
114
125
115
126
for file in iglob (filepath ):
116
- out .add ((file , regex ))
127
+ out .add ((file , re . compile ( regex ) ))
117
128
118
129
return sorted (out )
119
130
120
131
121
- def _bump_with_regex (
122
- version_filepath : str ,
123
- current_version : str ,
124
- new_version : str ,
125
- regex : str ,
126
- encoding : str = ENCODING ,
127
- ) -> tuple [bool , str ]:
132
+ def _get_bumped_version_file_content (
133
+ version_file : TextIO ,
134
+ bump_line : Callable [[str ], str ],
135
+ error_on_not_found : Callable [[], Never ] | None ,
136
+ ) -> str :
128
137
current_version_found = False
129
138
lines = []
130
- pattern = re .compile (regex )
131
- with open (version_filepath , encoding = encoding ) as f :
132
- for line in f :
133
- if not pattern .search (line ):
134
- lines .append (line )
135
- continue
136
-
137
- bumped_line = line .replace (current_version , new_version )
138
- if bumped_line != line :
139
- current_version_found = True
140
- lines .append (bumped_line )
141
-
142
- return current_version_found , "" .join (lines )
139
+
140
+ for line in version_file :
141
+ bumped_line = bump_line (line )
142
+ if bumped_line != line :
143
+ current_version_found = True
144
+
145
+ lines .append (bumped_line )
146
+
147
+ if error_on_not_found is not None and not current_version_found :
148
+ error_on_not_found ()
149
+
150
+ return "" .join (lines )
143
151
144
152
145
153
def create_commit_message (
0 commit comments