Coverage for utilities/data.py: 22%
133 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import decimal
2from itertools import count, groupby
4from django.db.backends.postgresql.psycopg_any import NumericRange
6__all__ = (
7 'array_to_ranges',
8 'array_to_string',
9 'check_ranges_overlap',
10 'deep_compare_dict',
11 'deepmerge',
12 'drange',
13 'flatten_dict',
14 'get_config_value_ci',
15 'get_inclusive_integer_range_bounds',
16 'normalize_integer_range',
17 'normalize_update_fields',
18 'ranges_to_string',
19 'ranges_to_string_list',
20 'resolve_attr_path',
21 'shallow_compare_dict',
22 'string_to_ranges',
23)
26#
27# Dictionary utilities
28#
30def get_config_value_ci(config_dict, key, default=None):
31 """
32 Retrieve a value from a dictionary using case-insensitive key matching.
33 """
34 if key in config_dict: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true
35 return config_dict[key]
36 key_lower = key.lower()
37 for config_key, value in config_dict.items(): 37 ↛ 38line 37 didn't jump to line 38 because the loop on line 37 never started
38 if config_key.lower() == key_lower:
39 return value
40 return default
43def deepmerge(original, new):
44 """
45 Deep merge two dictionaries (new into original) and return a new dict
46 """
47 merged = dict(original)
48 for key, val in new.items(): 48 ↛ 49line 48 didn't jump to line 49 because the loop on line 48 never started
49 if key in original and isinstance(original[key], dict) and val and isinstance(val, dict):
50 merged[key] = deepmerge(original[key], val)
51 else:
52 merged[key] = val
53 return merged
56def flatten_dict(d, prefix='', separator='.'):
57 """
58 Flatten nested dictionaries into a single level by joining key names with a separator.
60 :param d: The dictionary to be flattened
61 :param prefix: Initial prefix (if any)
62 :param separator: The character to use when concatenating key names
63 """
64 ret = {}
65 for k, v in d.items():
66 key = separator.join([prefix, k]) if prefix else k
67 if type(v) is dict:
68 ret.update(flatten_dict(v, prefix=key, separator=separator))
69 else:
70 ret[key] = v
71 return ret
74def shallow_compare_dict(source_dict, destination_dict, exclude=tuple()):
75 """
76 Return a new dictionary of the different keys. The values of `destination_dict` are returned. Only the equality of
77 the first layer of keys/values is checked. `exclude` is a list or tuple of keys to be ignored.
78 """
79 difference = {}
81 for key, value in destination_dict.items():
82 if key in exclude:
83 continue
84 if source_dict.get(key) != value:
85 difference[key] = value
87 return difference
90def deep_compare_dict(source_dict, destination_dict, exclude=tuple()):
91 """
92 Return a two-tuple of dictionaries (added, removed) representing the differences between source_dict and
93 destination_dict. For values which are themselves dicts, the comparison is performed recursively such that only
94 the changed keys within the nested dict are included. `exclude` is a list or tuple of keys to be ignored.
95 """
96 added = {}
97 removed = {}
99 all_keys = set(source_dict) | set(destination_dict)
100 for key in all_keys:
101 if key in exclude:
102 continue
103 src_val = source_dict.get(key)
104 dst_val = destination_dict.get(key)
105 if src_val == dst_val:
106 continue
107 if isinstance(src_val, dict) and isinstance(dst_val, dict):
108 sub_added, sub_removed = deep_compare_dict(src_val, dst_val)
109 if sub_added or sub_removed:
110 added[key] = sub_added
111 removed[key] = sub_removed
112 else:
113 added[key] = dst_val
114 removed[key] = src_val
116 return added, removed
119def normalize_update_fields(kwargs):
120 """
121 Replace `kwargs['update_fields']` with a frozenset and return it, so a save() override can
122 run membership tests without consuming a one-shot iterable. `None` and an absent key are
123 left alone.
124 """
125 update_fields = kwargs.get('update_fields')
126 if update_fields is not None: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 update_fields = frozenset(update_fields)
128 kwargs['update_fields'] = update_fields
129 return update_fields
132#
133# Array utilities
134#
136def array_to_ranges(array):
137 """
138 Convert an arbitrary array of integers to a list of consecutive values. Nonconsecutive values are returned as
139 single-item tuples.
141 Example:
142 [0, 1, 2, 10, 14, 15, 16] => [(0, 2), (10,), (14, 16)]
143 """
144 group = (
145 list(x) for _, x in groupby(sorted(array), lambda x, c=count(): next(c) - x)
146 )
147 return [
148 (g[0], g[-1])[:len(g)] for g in group
149 ]
152def array_to_string(array):
153 """
154 Generate an efficient, human-friendly string from a set of integers. Intended for use with ArrayField.
156 Example:
157 [0, 1, 2, 10, 14, 15, 16] => "0-2, 10, 14-16"
158 """
159 ret = []
160 ranges = array_to_ranges(array)
161 for value in ranges:
162 if len(value) == 1:
163 ret.append(str(value[0]))
164 else:
165 ret.append(f'{value[0]}-{value[1]}')
166 return ', '.join(ret)
169#
170# Range utilities
171#
173def drange(start, end, step=decimal.Decimal(1)):
174 """
175 Decimal-compatible implementation of Python's range()
176 """
177 start, end, step = decimal.Decimal(start), decimal.Decimal(end), decimal.Decimal(step)
178 if start < end:
179 while start < end:
180 yield start
181 start += step
182 else:
183 while start > end:
184 yield start
185 start += step
188def get_inclusive_integer_range_bounds(value_range):
189 """
190 Return the lower and upper bounds of a bounded, non-empty discrete
191 integer range as inclusive values.
193 For example, ``[10, 20)`` is returned as ``(10, 19)``, while
194 ``[10, 20]`` is returned as ``(10, 20)``.
196 Both bounds must be non-``None``; unbounded ranges are not supported.
197 """
198 lower = value_range.lower if value_range.lower_inc else value_range.lower + 1
199 upper = value_range.upper if value_range.upper_inc else value_range.upper - 1
201 return lower, upper
204def normalize_integer_range(value_range):
205 """
206 Return an equivalent canonical half-open ``[)`` range for a bounded,
207 non-empty discrete integer range, regardless of the input range's
208 bounds metadata.
209 """
210 lower, upper = get_inclusive_integer_range_bounds(value_range)
212 return NumericRange(lower, upper + 1, bounds='[)')
215def check_ranges_overlap(ranges):
216 """
217 Check for overlap in an iterable of NumericRanges. Does not mutate the input.
218 """
219 ranges = sorted(ranges, key=lambda value_range: get_inclusive_integer_range_bounds(value_range)[0])
221 for i in range(1, len(ranges)): 221 ↛ 222line 221 didn't jump to line 222 because the loop on line 221 never started
222 prev_upper = get_inclusive_integer_range_bounds(ranges[i - 1])[1]
223 lower = get_inclusive_integer_range_bounds(ranges[i])[0]
225 if prev_upper >= lower:
226 return True
228 return False
231def ranges_to_string_list(ranges):
232 """
233 Convert numeric ranges to a list of display strings.
235 Each range is rendered as "lower-upper" or "lower" (for singletons).
236 Bounds are normalized to inclusive values using ``lower_inc``/``upper_inc``.
237 This underpins ``ranges_to_string()``, which joins the result with commas.
239 Example:
240 [NumericRange(1, 6), NumericRange(8, 9), NumericRange(10, 13)] => ["1-5", "8", "10-12"]
241 """
242 if not ranges:
243 return []
245 output: list[str] = []
246 for r in ranges:
247 lower, upper = get_inclusive_integer_range_bounds(r)
248 output.append(f"{lower}-{upper}" if lower != upper else str(lower))
249 return output
252def ranges_to_string(ranges):
253 """
254 Converts a list of ranges into a string representation.
256 This function takes a list of range objects and produces a string
257 representation of those ranges. Each range is represented as a
258 hyphen-separated pair of lower and upper bounds, with inclusive or
259 exclusive bounds adjusted accordingly. If the lower and upper bounds
260 of a range are the same, only the single value is added to the string.
261 Intended for use with ArrayField.
263 Example:
264 [NumericRange(1, 5), NumericRange(8, 9), NumericRange(10, 12)] => "1-5,8,10-12"
265 """
266 if not ranges:
267 return ''
268 return ','.join(ranges_to_string_list(ranges))
271def string_to_ranges(value):
272 """
273 Converts a string representation of numeric ranges into a list of NumericRange objects.
275 This function parses a string containing numeric values and ranges separated by commas (e.g.,
276 "1-5,8,10-12") and converts it into a list of NumericRange objects.
277 In the case of a single integer, it is treated as a range where the start and end
278 are equal. The returned ranges are represented as half-open intervals [lower, upper).
279 Intended for use with ArrayField.
281 Example:
282 "1-5,8,10-12" => [NumericRange(1, 6), NumericRange(8, 9), NumericRange(10, 13)]
283 """
284 if not value:
285 return None
286 value.replace(' ', '') # Remove whitespace
287 values = []
288 for data in value.split(','):
289 dash_range = data.strip().split('-')
290 if len(dash_range) == 1 and str(dash_range[0]).isdigit():
291 # Single integer value; expand to a range
292 lower = dash_range[0]
293 upper = dash_range[0]
294 elif len(dash_range) == 2 and str(dash_range[0]).isdigit() and str(dash_range[1]).isdigit():
295 # The range has two values and both are valid integers
296 lower = dash_range[0]
297 upper = dash_range[1]
298 else:
299 return None
300 values.append(NumericRange(int(lower), int(upper) + 1, bounds='[)'))
301 return values
304#
305# Attribute resolution
306#
308def resolve_attr_path(obj, path):
309 """
310 Follow a dotted path across attributes and/or dictionary keys and return the final value.
312 Parameters:
313 obj: The starting object
314 path: The dotted path to follow (e.g. "foo.bar.baz")
315 """
316 cur = obj
317 for part in path.split('.'):
318 if cur is None:
319 return None
320 try:
321 cur = getattr(cur, part) if hasattr(cur, part) else cur.get(part)
322 except AttributeError:
323 cur = None
324 return cur