Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/black/ranges.py: 14%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Functions related to Black's formatting by line ranges feature."""
3import difflib
4from collections.abc import Collection, Iterator, Sequence
5from dataclasses import dataclass
7from black.nodes import (
8 LN,
9 STANDALONE_COMMENT,
10 Leaf,
11 Node,
12 Visitor,
13 first_leaf,
14 furthest_ancestor_with_last_leaf,
15 last_leaf,
16 syms,
17)
18from blib2to3.pgen2.token import ASYNC, NEWLINE
21def parse_line_ranges(line_ranges: Sequence[str]) -> list[tuple[int, int]]:
22 lines: list[tuple[int, int]] = []
23 for lines_str in line_ranges:
24 parts = lines_str.split("-")
25 if len(parts) != 2:
26 raise ValueError(
27 "Incorrect --line-ranges format, expect 'START-END', found"
28 f" {lines_str!r}"
29 )
30 try:
31 start = int(parts[0])
32 end = int(parts[1])
33 except ValueError:
34 raise ValueError(
35 "Incorrect --line-ranges value, expect integer ranges, found"
36 f" {lines_str!r}"
37 ) from None
38 else:
39 if start < 1 or end < 1:
40 raise ValueError(
41 "Incorrect --line-ranges value, "
42 f"expect positive integers, found {lines_str!r}"
43 )
44 if start > end:
45 raise ValueError(
46 "Incorrect --line-ranges value, expect START <= END, "
47 f"found {lines_str!r}"
48 )
49 lines.append((start, end))
50 return lines
53def is_valid_line_range(lines: tuple[int, int]) -> bool:
54 """Returns whether the line range is valid."""
55 return not lines or lines[0] <= lines[1]
58def sanitized_lines(
59 lines: Collection[tuple[int, int]], src_contents: str
60) -> Collection[tuple[int, int]]:
61 """Returns the valid line ranges for the given source.
63 This removes ranges that are entirely outside the valid lines.
65 Other ranges are normalized so that the start values are at least 1 and the
66 end values are at most the (1-based) index of the last source line.
67 """
68 if not src_contents:
69 return []
70 good_lines = []
71 src_line_count = src_contents.count("\n")
72 if not src_contents.endswith("\n"):
73 src_line_count += 1
74 for start, end in lines:
75 if start > src_line_count:
76 continue
77 # line-ranges are 1-based
78 start = max(start, 1)
79 if end < start:
80 continue
81 end = min(end, src_line_count)
82 good_lines.append((start, end))
83 return good_lines
86def adjusted_lines(
87 lines: Collection[tuple[int, int]],
88 original_source: str,
89 modified_source: str,
90) -> list[tuple[int, int]]:
91 """Returns the adjusted line ranges based on edits from the original code.
93 This computes the new line ranges by diffing original_source and
94 modified_source, and adjust each range based on how the range overlaps with
95 the diffs.
97 Note the diff can contain lines outside of the original line ranges. This can
98 happen when the formatting has to be done in adjacent to maintain consistent
99 local results. For example:
101 1. def my_func(arg1, arg2,
102 2. arg3,):
103 3. pass
105 If it restricts to line 2-2, it can't simply reformat line 2, it also has
106 to reformat line 1:
108 1. def my_func(
109 2. arg1,
110 3. arg2,
111 4. arg3,
112 5. ):
113 6. pass
115 In this case, we will expand the line ranges to also include the whole diff
116 block.
118 Args:
119 lines: a collection of line ranges.
120 original_source: the original source.
121 modified_source: the modified source.
122 """
123 lines_mappings = _calculate_lines_mappings(original_source, modified_source)
125 new_lines = []
126 # Keep an index of the current search. Since the lines and lines_mappings are
127 # sorted, this makes the search complexity linear.
128 current_mapping_index = 0
129 for start, end in sorted(lines):
130 start_mapping_index = _find_lines_mapping_index(
131 start,
132 lines_mappings,
133 current_mapping_index,
134 )
135 end_mapping_index = _find_lines_mapping_index(
136 end,
137 lines_mappings,
138 start_mapping_index,
139 )
140 current_mapping_index = start_mapping_index
141 if start_mapping_index >= len(lines_mappings) or end_mapping_index >= len(
142 lines_mappings
143 ):
144 # Protect against invalid inputs.
145 continue
146 start_mapping = lines_mappings[start_mapping_index]
147 end_mapping = lines_mappings[end_mapping_index]
148 if start_mapping.is_changed_block:
149 # When the line falls into a changed block, expands to the whole block.
150 new_start = start_mapping.modified_start
151 else:
152 new_start = (
153 start - start_mapping.original_start + start_mapping.modified_start
154 )
155 if end_mapping.is_changed_block:
156 # When the line falls into a changed block, expands to the whole block.
157 new_end = end_mapping.modified_end
158 else:
159 new_end = end - end_mapping.original_start + end_mapping.modified_start
160 new_range = (new_start, new_end)
161 if is_valid_line_range(new_range):
162 new_lines.append(new_range)
163 return new_lines
166def convert_unchanged_lines(src_node: Node, lines: Collection[tuple[int, int]]) -> None:
167 r"""Converts unchanged lines to STANDALONE_COMMENT.
169 The idea is similar to how `# fmt: on/off` is implemented. It also converts the
170 nodes between those markers as a single `STANDALONE_COMMENT` leaf node with
171 the unformatted code as its value. `STANDALONE_COMMENT` is a "fake" token
172 that will be formatted as-is with its prefix normalized.
174 Here we perform two passes:
176 1. Visit the top-level statements, and convert them to a single
177 `STANDALONE_COMMENT` when unchanged. This speeds up formatting when some
178 of the top-level statements aren't changed.
179 2. Convert unchanged "unwrapped lines" to `STANDALONE_COMMENT` nodes line by
180 line. "unwrapped lines" are divided by the `NEWLINE` token. e.g. a
181 multi-line statement is *one* "unwrapped line" that ends with `NEWLINE`,
182 even though this statement itself can span multiple lines, and the
183 tokenizer only sees the last '\n' as the `NEWLINE` token.
185 NOTE: During pass (2), comment prefixes and indentations are ALWAYS
186 normalized even when the lines aren't changed. This is fixable by moving
187 more formatting to pass (1). However, it's hard to get it correct when
188 incorrect indentations are used. So we defer this to future optimizations.
189 """
190 lines_set: set[int] = set()
191 for start, end in lines:
192 lines_set.update(range(start, end + 1))
193 replacements = _NodeReplacements()
194 visitor = _TopLevelStatementsVisitor(lines_set, replacements)
195 _ = list(visitor.visit(src_node)) # Consume all results.
196 replacements.apply()
197 _convert_unchanged_line_by_line(src_node, lines_set)
200class _NodeReplacements:
201 """Collects STANDALONE_COMMENT conversions and applies them per parent.
203 Converting a node splices a `STANDALONE_COMMENT` leaf in place of a run of
204 sibling nodes. Doing that one node at a time with `Base.remove` /
205 `insert_child` scans and shifts the parent's children list on every call, so
206 converting the many sibling blocks of one parent (a long if/elif chain, a
207 match with many cases, a module with many top-level statements) is O(n^2).
208 Recording the conversions and rewriting each parent's children in a single
209 pass makes it O(n).
210 """
212 def __init__(self) -> None:
213 # id(parent) -> (parent, {id(child): standalone to splice in}, {ids to drop})
214 # A run's nodes can live under more than one parent (an `async` statement
215 # keeps the `ASYNC` leaf on the grandparent), so the standalone replaces
216 # the run's first node under its own parent while the remaining nodes are
217 # just dropped from wherever they sit.
218 self._by_parent: dict[int, tuple[Node, dict[int, Leaf], set[int]]] = {}
219 # NEWLINE leaves already inside a recorded run, so the line-by-line pass
220 # doesn't record an overlapping conversion for them.
221 self.covered_newlines: set[int] = set()
222 # Nodes already scheduled for removal. When conversions immediately
223 # mutated the tree, a second conversion touching an already-removed node
224 # was a no-op (`Base.remove` returned None); recording defers the
225 # mutation, so we skip such nodes explicitly to keep that behaviour.
226 self._recorded: set[int] = set()
227 # A leaf's prefix (indentation, blank lines, comments) moves onto its
228 # STANDALONE_COMMENT. When an inner and an outer node that share the same
229 # first leaf are both converted (a decorator inside a decorated block),
230 # the earlier conversion clears the prefix off the leaf, so the later one
231 # would see an empty prefix. Remember the taken prefix keyed by the leaf
232 # so both conversions reuse the original.
233 self._taken_prefixes: dict[int, str] = {}
235 def _entry(self, parent: Node) -> tuple[Node, dict[int, Leaf], set[int]]:
236 return self._by_parent.setdefault(id(parent), (parent, {}, set()))
238 def take_prefix(self, first: Leaf) -> str:
239 cached = self._taken_prefixes.get(id(first))
240 if cached is not None:
241 return cached
242 prefix = first.prefix
243 first.prefix = ""
244 # Immediate conversion replaced the whole run with a freshly built
245 # STANDALONE_COMMENT leaf, whose synthesized position is line 0. Recording
246 # the conversion leaves the original leaf in place, so mirror that line 0
247 # here: when a later, enclosing node shares this first leaf (a decorated
248 # block over an already-recorded decorator), _get_line_range sees the run
249 # as starting at line 0, matching the immediate behaviour.
250 first.lineno = 0
251 self._taken_prefixes[id(first)] = prefix
252 return prefix
254 def is_recorded(self, node: LN) -> bool:
255 return id(node) in self._recorded
257 def record(self, run: list[LN], standalone: Leaf) -> None:
258 first = run[0]
259 first_parent = first.parent
260 if first_parent is None or id(first) in self._recorded:
261 return
262 for node in run:
263 self._recorded.add(id(node))
264 for leaf in node.leaves():
265 if leaf.type == NEWLINE:
266 self.covered_newlines.add(id(leaf))
267 parent = node.parent
268 if parent is not None:
269 self._entry(parent)[2].add(id(node))
270 self._entry(first_parent)[1][id(first)] = standalone
272 def apply(self) -> None:
273 for parent, replaced, removed in self._by_parent.values():
274 new_children: list[LN] = []
275 for child in parent.children:
276 standalone = replaced.get(id(child))
277 if standalone is not None:
278 standalone.parent = parent
279 new_children.append(standalone)
280 child.parent = None
281 elif id(child) in removed:
282 child.parent = None
283 else:
284 new_children.append(child)
285 parent.children = new_children
286 parent.changed()
287 parent.invalidate_sibling_maps()
290def _contains_standalone_comment(node: LN) -> bool:
291 if isinstance(node, Leaf):
292 return node.type == STANDALONE_COMMENT
293 else:
294 for child in node.children:
295 if _contains_standalone_comment(child):
296 return True
297 return False
300class _TopLevelStatementsVisitor(Visitor[None]):
301 """
302 A node visitor that converts unchanged top-level statements to
303 STANDALONE_COMMENT.
305 This is used in addition to _convert_unchanged_line_by_line, to
306 speed up formatting when there are unchanged top-level
307 classes/functions/statements.
308 """
310 def __init__(self, lines_set: set[int], replacements: "_NodeReplacements"):
311 self._lines_set = lines_set
312 self._replacements = replacements
314 def visit_simple_stmt(self, node: Node) -> Iterator[None]:
315 # This is only called for top-level statements, since `visit_suite`
316 # won't visit its children nodes.
317 yield from []
318 newline_leaf = last_leaf(node)
319 if not newline_leaf:
320 return
321 assert (
322 newline_leaf.type == NEWLINE
323 ), f"Unexpectedly found leaf.type={newline_leaf.type}"
324 # We need to find the furthest ancestor with the NEWLINE as the last
325 # leaf, since a `suite` can simply be a `simple_stmt` when it puts
326 # its body on the same line. Example: `if cond: pass`.
327 ancestor = furthest_ancestor_with_last_leaf(newline_leaf)
328 if not _get_line_range(ancestor).intersection(self._lines_set):
329 _convert_node_to_standalone_comment(ancestor, self._replacements)
331 def visit_suite(self, node: Node) -> Iterator[None]:
332 yield from []
333 # If there is a STANDALONE_COMMENT node, it means parts of the node tree
334 # have fmt on/off/skip markers. Those STANDALONE_COMMENT nodes can't
335 # be simply converted by calling str(node). So we just don't convert
336 # here.
337 if _contains_standalone_comment(node):
338 return
339 # Find the semantic parent of this suite. For `async_stmt` and
340 # `async_funcdef`, the ASYNC token is defined on a separate level by the
341 # grammar.
342 semantic_parent = node.parent
343 if semantic_parent is not None:
344 if (
345 semantic_parent.prev_sibling is not None
346 and semantic_parent.prev_sibling.type == ASYNC
347 ):
348 semantic_parent = semantic_parent.parent
349 if semantic_parent is not None and not _get_line_range(
350 semantic_parent
351 ).intersection(self._lines_set):
352 _convert_node_to_standalone_comment(semantic_parent, self._replacements)
355def _convert_unchanged_line_by_line(node: Node, lines_set: set[int]) -> None:
356 """Converts unchanged to STANDALONE_COMMENT line by line."""
357 replacements = _NodeReplacements()
358 for leaf in node.leaves():
359 if leaf.type != NEWLINE:
360 # We only consider "unwrapped lines", which are divided by the NEWLINE
361 # token.
362 continue
363 if id(leaf) in replacements.covered_newlines:
364 # This NEWLINE is inside a run already scheduled for conversion (e.g.
365 # a second decorator on a stacked decorator block).
366 continue
367 if leaf.parent and leaf.parent.type == syms.match_stmt:
368 # The `suite` node is defined as:
369 # match_stmt: "match" subject_expr ':' NEWLINE INDENT case_block+ DEDENT
370 # Here we need to check `subject_expr`. The `case_block+` will be
371 # checked by their own NEWLINEs.
372 nodes_to_ignore: list[LN] = []
373 prev_sibling = leaf.prev_sibling
374 while prev_sibling:
375 nodes_to_ignore.insert(0, prev_sibling)
376 prev_sibling = prev_sibling.prev_sibling
377 if not _get_line_range(nodes_to_ignore).intersection(lines_set):
378 _convert_nodes_to_standalone_comment(
379 nodes_to_ignore, newline=leaf, replacements=replacements
380 )
381 elif leaf.parent and leaf.parent.type == syms.suite:
382 # The `suite` node is defined as:
383 # suite: simple_stmt | NEWLINE INDENT stmt+ DEDENT
384 # We will check `simple_stmt` and `stmt+` separately against the lines set
385 parent_sibling = leaf.parent.prev_sibling
386 nodes_to_ignore = []
387 while parent_sibling and parent_sibling.type != syms.suite:
388 # NOTE: Multiple suite nodes can exist as siblings in e.g. `if_stmt`.
389 nodes_to_ignore.insert(0, parent_sibling)
390 parent_sibling = parent_sibling.prev_sibling
391 # Special case for `async_stmt` and `async_funcdef` where the ASYNC
392 # token is on the grandparent node.
393 grandparent = leaf.parent.parent
394 if (
395 grandparent is not None
396 and grandparent.prev_sibling is not None
397 and grandparent.prev_sibling.type == ASYNC
398 ):
399 nodes_to_ignore.insert(0, grandparent.prev_sibling)
400 if not _get_line_range(nodes_to_ignore).intersection(lines_set):
401 _convert_nodes_to_standalone_comment(
402 nodes_to_ignore, newline=leaf, replacements=replacements
403 )
404 else:
405 ancestor = furthest_ancestor_with_last_leaf(leaf)
406 # Consider multiple decorators as a whole block, as their
407 # newlines have different behaviors than the rest of the grammar.
408 if (
409 ancestor.type == syms.decorator
410 and ancestor.parent
411 and ancestor.parent.type == syms.decorators
412 ):
413 ancestor = ancestor.parent
414 if not _get_line_range(ancestor).intersection(lines_set):
415 _convert_node_to_standalone_comment(ancestor, replacements)
416 replacements.apply()
419def _convert_node_to_standalone_comment(
420 node: LN, replacements: "_NodeReplacements"
421) -> None:
422 """Convert node to STANDALONE_COMMENT by modifying the tree inline."""
423 parent = node.parent
424 if not parent or replacements.is_recorded(node):
425 return
426 first = first_leaf(node)
427 last = last_leaf(node)
428 if not first or not last:
429 return
430 if first is last:
431 # This can happen on the following edge cases:
432 # 1. A block of `# fmt: off/on` code except the `# fmt: on` is placed
433 # on the end of the last line instead of on a new line.
434 # 2. A single backslash on its own line followed by a comment line.
435 # Ideally we don't want to format them when not requested, but fixing
436 # isn't easy. These cases are also badly formatted code, so it isn't
437 # too bad we reformat them.
438 return
439 # The prefix contains comments and indentation whitespaces. They are
440 # reformatted accordingly to the correct indentation level.
441 # This also means the indentation will be changed on the unchanged lines, and
442 # this is actually required to not break incremental reformatting.
443 prefix = replacements.take_prefix(first)
444 # For a single-line decorated item the decorator and the item need a newline
445 # between them. The conversions are recorded and applied in a single pass
446 # instead of mutating the tree per node, so the decorator here is still its
447 # own child node carrying its trailing NEWLINE, which already separates it
448 # from the item; there's nothing to add back. (A decorated item spanning
449 # multiple lines is handled by the earlier suite case in
450 # _convert_unchanged_line_by_line, which manages the newlines itself.)
451 # Remove the '\n', as STANDALONE_COMMENT will have '\n' appended when
452 # generating the formatted code.
453 value = str(node)[:-1]
454 replacements.record(
455 [node],
456 Leaf(
457 STANDALONE_COMMENT,
458 value,
459 prefix=prefix,
460 fmt_pass_converted_first_leaf=first,
461 ),
462 )
465def _convert_nodes_to_standalone_comment(
466 nodes: Sequence[LN], *, newline: Leaf, replacements: "_NodeReplacements"
467) -> None:
468 """Convert nodes to STANDALONE_COMMENT by modifying the tree inline."""
469 if not nodes:
470 return
471 parent = nodes[0].parent
472 first = first_leaf(nodes[0])
473 if not parent or not first or replacements.is_recorded(nodes[0]):
474 return
475 prefix = replacements.take_prefix(first)
476 value = "".join(str(node) for node in nodes)
477 # The prefix comment on the NEWLINE leaf is the trailing comment of the statement.
478 if newline.prefix:
479 value += newline.prefix
480 newline.prefix = ""
481 replacements.record(
482 list(nodes),
483 Leaf(
484 STANDALONE_COMMENT,
485 value,
486 prefix=prefix,
487 fmt_pass_converted_first_leaf=first,
488 ),
489 )
492def _leaf_line_end(leaf: Leaf) -> int:
493 """Returns the line number of the leaf node's last line."""
494 if leaf.type == NEWLINE:
495 return leaf.lineno
496 else:
497 # Leaf nodes like multiline strings can occupy multiple lines.
498 return leaf.lineno + str(leaf).count("\n")
501def _get_line_range(node_or_nodes: LN | list[LN]) -> set[int]:
502 """Returns the line range of this node or list of nodes."""
503 if isinstance(node_or_nodes, list):
504 nodes = node_or_nodes
505 if not nodes:
506 return set()
507 first = first_leaf(nodes[0])
508 last = last_leaf(nodes[-1])
509 if first and last:
510 line_start = first.lineno
511 line_end = _leaf_line_end(last)
512 return set(range(line_start, line_end + 1))
513 else:
514 return set()
515 else:
516 node = node_or_nodes
517 if isinstance(node, Leaf):
518 return set(range(node.lineno, _leaf_line_end(node) + 1))
519 else:
520 first = first_leaf(node)
521 last = last_leaf(node)
522 if first and last:
523 return set(range(first.lineno, _leaf_line_end(last) + 1))
524 else:
525 return set()
528@dataclass
529class _LinesMapping:
530 """1-based lines mapping from original source to modified source.
532 Lines [original_start, original_end] from original source
533 are mapped to [modified_start, modified_end].
535 The ranges are inclusive on both ends.
536 """
538 original_start: int
539 original_end: int
540 modified_start: int
541 modified_end: int
542 # Whether this range corresponds to a changed block, or an unchanged block.
543 is_changed_block: bool
546def _calculate_lines_mappings(
547 original_source: str,
548 modified_source: str,
549) -> Sequence[_LinesMapping]:
550 """Returns a sequence of _LinesMapping by diffing the sources.
552 For example, given the following diff:
553 import re
554 - def func(arg1,
555 - arg2, arg3):
556 + def func(arg1, arg2, arg3):
557 pass
558 It returns the following mappings:
559 original -> modified
560 (1, 1) -> (1, 1), is_changed_block=False (the "import re" line)
561 (2, 3) -> (2, 2), is_changed_block=True (the diff)
562 (4, 4) -> (3, 3), is_changed_block=False (the "pass" line)
564 You can think of this visually as if it brings up a side-by-side diff, and tries
565 to map the line ranges from the left side to the right side:
567 (1, 1)->(1, 1) 1. import re 1. import re
568 (2, 3)->(2, 2) 2. def func(arg1, 2. def func(arg1, arg2, arg3):
569 3. arg2, arg3):
570 (4, 4)->(3, 3) 4. pass 3. pass
572 Args:
573 original_source: the original source.
574 modified_source: the modified source.
575 """
576 matcher = difflib.SequenceMatcher(
577 None,
578 original_source.splitlines(keepends=True),
579 modified_source.splitlines(keepends=True),
580 )
581 matching_blocks = matcher.get_matching_blocks()
582 lines_mappings: list[_LinesMapping] = []
583 # matching_blocks is a sequence of "same block of code ranges", see
584 # https://docs.python.org/3/library/difflib.html#difflib.SequenceMatcher.get_matching_blocks
585 # Each block corresponds to a _LinesMapping with is_changed_block=False,
586 # and the ranges between two blocks corresponds to a _LinesMapping with
587 # is_changed_block=True,
588 # NOTE: matching_blocks is 0-based, but _LinesMapping is 1-based.
589 for i, block in enumerate(matching_blocks):
590 if i == 0:
591 if block.a != 0 or block.b != 0:
592 lines_mappings.append(
593 _LinesMapping(
594 original_start=1,
595 original_end=block.a,
596 modified_start=1,
597 modified_end=block.b,
598 is_changed_block=False,
599 )
600 )
601 else:
602 previous_block = matching_blocks[i - 1]
603 lines_mappings.append(
604 _LinesMapping(
605 original_start=previous_block.a + previous_block.size + 1,
606 original_end=block.a,
607 modified_start=previous_block.b + previous_block.size + 1,
608 modified_end=block.b,
609 is_changed_block=True,
610 )
611 )
612 if i < len(matching_blocks) - 1:
613 lines_mappings.append(
614 _LinesMapping(
615 original_start=block.a + 1,
616 original_end=block.a + block.size,
617 modified_start=block.b + 1,
618 modified_end=block.b + block.size,
619 is_changed_block=False,
620 )
621 )
622 return lines_mappings
625def _find_lines_mapping_index(
626 original_line: int,
627 lines_mappings: Sequence[_LinesMapping],
628 start_index: int,
629) -> int:
630 """Returns the original index of the lines mappings for the original line."""
631 index = start_index
632 while index < len(lines_mappings):
633 mapping = lines_mappings[index]
634 if mapping.original_start <= original_line <= mapping.original_end:
635 return index
636 index += 1
637 return index