All patches and comments are welcome. Please squash your changes to logical
commits before using git-format-patch and git-send-email to
patches@git.madduck.net.
If you'd read over the Git project's submission guidelines and adhered to them,
I'd be especially grateful.
2 from concurrent.futures import Executor, ProcessPoolExecutor
3 from contextlib import contextmanager
4 from datetime import datetime
6 from functools import lru_cache, partial, wraps
10 from multiprocessing import Manager, freeze_support
12 from pathlib import Path
40 from appdirs import user_cache_dir
41 from attr import dataclass, evolve, Factory
44 from typed_ast import ast3, ast27
47 from blib2to3.pytree import Node, Leaf, type_repr
48 from blib2to3 import pygram, pytree
49 from blib2to3.pgen2 import driver, token
50 from blib2to3.pgen2.grammar import Grammar
51 from blib2to3.pgen2.parse import ParseError
54 __version__ = "19.3b0"
55 DEFAULT_LINE_LENGTH = 88
57 r"/(\.eggs|\.git|\.hg|\.mypy_cache|\.nox|\.tox|\.venv|_build|buck-out|build|dist)/"
59 DEFAULT_INCLUDES = r"\.pyi?$"
60 CACHE_DIR = Path(user_cache_dir("black", version=__version__))
72 LN = Union[Leaf, Node]
73 SplitFunc = Callable[["Line", Collection["Feature"]], Iterator["Line"]]
76 CacheInfo = Tuple[Timestamp, FileSize]
77 Cache = Dict[Path, CacheInfo]
78 out = partial(click.secho, bold=True, err=True)
79 err = partial(click.secho, fg="red", err=True)
81 pygram.initialize(CACHE_DIR)
82 syms = pygram.python_symbols
85 class NothingChanged(UserWarning):
86 """Raised when reformatted code is the same as source."""
89 class CannotSplit(Exception):
90 """A readable split that fits the allotted line length is impossible."""
93 class InvalidInput(ValueError):
94 """Raised when input source code fails all parse attempts."""
97 class WriteBack(Enum):
104 def from_configuration(cls, *, check: bool, diff: bool) -> "WriteBack":
105 if check and not diff:
108 return cls.DIFF if diff else cls.YES
117 class TargetVersion(Enum):
126 def is_python2(self) -> bool:
127 return self is TargetVersion.PY27
130 PY36_VERSIONS = {TargetVersion.PY36, TargetVersion.PY37, TargetVersion.PY38}
134 # All string literals are unicode
137 NUMERIC_UNDERSCORES = 3
138 TRAILING_COMMA_IN_CALL = 4
139 TRAILING_COMMA_IN_DEF = 5
140 # The following two feature-flags are mutually exclusive, and exactly one should be
141 # set for every version of python.
142 ASYNC_IDENTIFIERS = 6
146 VERSION_TO_FEATURES: Dict[TargetVersion, Set[Feature]] = {
147 TargetVersion.PY27: {Feature.ASYNC_IDENTIFIERS},
148 TargetVersion.PY33: {Feature.UNICODE_LITERALS, Feature.ASYNC_IDENTIFIERS},
149 TargetVersion.PY34: {Feature.UNICODE_LITERALS, Feature.ASYNC_IDENTIFIERS},
150 TargetVersion.PY35: {
151 Feature.UNICODE_LITERALS,
152 Feature.TRAILING_COMMA_IN_CALL,
153 Feature.ASYNC_IDENTIFIERS,
155 TargetVersion.PY36: {
156 Feature.UNICODE_LITERALS,
158 Feature.NUMERIC_UNDERSCORES,
159 Feature.TRAILING_COMMA_IN_CALL,
160 Feature.TRAILING_COMMA_IN_DEF,
161 Feature.ASYNC_IDENTIFIERS,
163 TargetVersion.PY37: {
164 Feature.UNICODE_LITERALS,
166 Feature.NUMERIC_UNDERSCORES,
167 Feature.TRAILING_COMMA_IN_CALL,
168 Feature.TRAILING_COMMA_IN_DEF,
169 Feature.ASYNC_KEYWORDS,
171 TargetVersion.PY38: {
172 Feature.UNICODE_LITERALS,
174 Feature.NUMERIC_UNDERSCORES,
175 Feature.TRAILING_COMMA_IN_CALL,
176 Feature.TRAILING_COMMA_IN_DEF,
177 Feature.ASYNC_KEYWORDS,
184 target_versions: Set[TargetVersion] = Factory(set)
185 line_length: int = DEFAULT_LINE_LENGTH
186 string_normalization: bool = True
189 def get_cache_key(self) -> str:
190 if self.target_versions:
191 version_str = ",".join(
193 for version in sorted(self.target_versions, key=lambda v: v.value)
199 str(self.line_length),
200 str(int(self.string_normalization)),
201 str(int(self.is_pyi)),
203 return ".".join(parts)
206 def supports_feature(target_versions: Set[TargetVersion], feature: Feature) -> bool:
207 return all(feature in VERSION_TO_FEATURES[version] for version in target_versions)
210 def read_pyproject_toml(
211 ctx: click.Context, param: click.Parameter, value: Union[str, int, bool, None]
213 """Inject Black configuration from "pyproject.toml" into defaults in `ctx`.
215 Returns the path to a successfully found and read configuration file, None
218 assert not isinstance(value, (int, bool)), "Invalid parameter type passed"
220 root = find_project_root(ctx.params.get("src", ()))
221 path = root / "pyproject.toml"
228 pyproject_toml = toml.load(value)
229 config = pyproject_toml.get("tool", {}).get("black", {})
230 except (toml.TomlDecodeError, OSError) as e:
231 raise click.FileError(
232 filename=value, hint=f"Error reading configuration file: {e}"
238 if ctx.default_map is None:
240 ctx.default_map.update( # type: ignore # bad types in .pyi
241 {k.replace("--", "").replace("-", "_"): v for k, v in config.items()}
246 @click.command(context_settings=dict(help_option_names=["-h", "--help"]))
247 @click.option("-c", "--code", type=str, help="Format the code passed in as a string.")
252 default=DEFAULT_LINE_LENGTH,
253 help="How many characters per line to allow.",
259 type=click.Choice([v.name.lower() for v in TargetVersion]),
260 callback=lambda c, p, v: [TargetVersion[val.upper()] for val in v],
263 "Python versions that should be supported by Black's output. [default: "
264 "per-file auto-detection]"
271 "Allow using Python 3.6-only syntax on all input files. This will put "
272 "trailing commas in function signatures and calls also after *args and "
273 "**kwargs. Deprecated; use --target-version instead. "
274 "[default: per-file auto-detection]"
281 "Format all input files like typing stubs regardless of file extension "
282 "(useful when piping source on standard input)."
287 "--skip-string-normalization",
289 help="Don't normalize string quotes or prefixes.",
295 "Don't write the files back, just return the status. Return code 0 "
296 "means nothing would change. Return code 1 means some files would be "
297 "reformatted. Return code 123 means there was an internal error."
303 help="Don't write the files back, just output a diff for each file on stdout.",
308 help="If --fast given, skip temporary sanity checks. [default: --safe]",
313 default=DEFAULT_INCLUDES,
315 "A regular expression that matches files and directories that should be "
316 "included on recursive searches. An empty value means all files are "
317 "included regardless of the name. Use forward slashes for directories on "
318 "all platforms (Windows, too). Exclusions are calculated first, inclusions "
326 default=DEFAULT_EXCLUDES,
328 "A regular expression that matches files and directories that should be "
329 "excluded on recursive searches. An empty value means no paths are excluded. "
330 "Use forward slashes for directories on all platforms (Windows, too). "
331 "Exclusions are calculated first, inclusions later."
340 "Don't emit non-error messages to stderr. Errors are still emitted; "
341 "silence those with 2>/dev/null."
349 "Also emit messages to stderr about files that were not changed or were "
350 "ignored due to --exclude=."
353 @click.version_option(version=__version__)
358 exists=True, file_okay=True, dir_okay=True, readable=True, allow_dash=True
365 exists=False, file_okay=True, dir_okay=False, readable=True, allow_dash=False
368 callback=read_pyproject_toml,
369 help="Read configuration from PATH.",
376 target_version: List[TargetVersion],
382 skip_string_normalization: bool,
388 config: Optional[str],
390 """The uncompromising code formatter."""
391 write_back = WriteBack.from_configuration(check=check, diff=diff)
394 err(f"Cannot use both --target-version and --py36")
397 versions = set(target_version)
400 "--py36 is deprecated and will be removed in a future version. "
401 "Use --target-version py36 instead."
403 versions = PY36_VERSIONS
405 # We'll autodetect later.
408 target_versions=versions,
409 line_length=line_length,
411 string_normalization=not skip_string_normalization,
413 if config and verbose:
414 out(f"Using configuration from {config}.", bold=False, fg="blue")
416 print(format_str(code, mode=mode))
419 include_regex = re_compile_maybe_verbose(include)
421 err(f"Invalid regular expression for include given: {include!r}")
424 exclude_regex = re_compile_maybe_verbose(exclude)
426 err(f"Invalid regular expression for exclude given: {exclude!r}")
428 report = Report(check=check, quiet=quiet, verbose=verbose)
429 root = find_project_root(src)
430 sources: Set[Path] = set()
435 gen_python_files_in_dir(p, root, include_regex, exclude_regex, report)
437 elif p.is_file() or s == "-":
438 # if a file was explicitly given, we don't care about its extension
441 err(f"invalid path: {s}")
442 if len(sources) == 0:
443 if verbose or not quiet:
444 out("No paths given. Nothing to do 😴")
447 if len(sources) == 1:
451 write_back=write_back,
457 sources=sources, fast=fast, write_back=write_back, mode=mode, report=report
460 if verbose or not quiet:
461 out("Oh no! 💥 💔 💥" if report.return_code else "All done! ✨ 🍰 ✨")
462 click.secho(str(report), err=True)
463 ctx.exit(report.return_code)
467 src: Path, fast: bool, write_back: WriteBack, mode: FileMode, report: "Report"
469 """Reformat a single file under `src` without spawning child processes.
471 `fast`, `write_back`, and `mode` options are passed to
472 :func:`format_file_in_place` or :func:`format_stdin_to_stdout`.
476 if not src.is_file() and str(src) == "-":
477 if format_stdin_to_stdout(fast=fast, write_back=write_back, mode=mode):
478 changed = Changed.YES
481 if write_back != WriteBack.DIFF:
482 cache = read_cache(mode)
483 res_src = src.resolve()
484 if res_src in cache and cache[res_src] == get_cache_info(res_src):
485 changed = Changed.CACHED
486 if changed is not Changed.CACHED and format_file_in_place(
487 src, fast=fast, write_back=write_back, mode=mode
489 changed = Changed.YES
490 if (write_back is WriteBack.YES and changed is not Changed.CACHED) or (
491 write_back is WriteBack.CHECK and changed is Changed.NO
493 write_cache(cache, [src], mode)
494 report.done(src, changed)
495 except Exception as exc:
496 report.failed(src, str(exc))
502 write_back: WriteBack,
506 """Reformat multiple files using a ProcessPoolExecutor."""
507 loop = asyncio.get_event_loop()
508 worker_count = os.cpu_count()
509 if sys.platform == "win32":
510 # Work around https://bugs.python.org/issue26903
511 worker_count = min(worker_count, 61)
512 executor = ProcessPoolExecutor(max_workers=worker_count)
514 loop.run_until_complete(
518 write_back=write_back,
530 async def schedule_formatting(
533 write_back: WriteBack,
536 loop: asyncio.AbstractEventLoop,
539 """Run formatting of `sources` in parallel using the provided `executor`.
541 (Use ProcessPoolExecutors for actual parallelism.)
543 `write_back`, `fast`, and `mode` options are passed to
544 :func:`format_file_in_place`.
547 if write_back != WriteBack.DIFF:
548 cache = read_cache(mode)
549 sources, cached = filter_cached(cache, sources)
550 for src in sorted(cached):
551 report.done(src, Changed.CACHED)
556 sources_to_cache = []
558 if write_back == WriteBack.DIFF:
559 # For diff output, we need locks to ensure we don't interleave output
560 # from different processes.
562 lock = manager.Lock()
564 asyncio.ensure_future(
565 loop.run_in_executor(
566 executor, format_file_in_place, src, fast, mode, write_back, lock
569 for src in sorted(sources)
571 pending: Iterable[asyncio.Future] = tasks.keys()
573 loop.add_signal_handler(signal.SIGINT, cancel, pending)
574 loop.add_signal_handler(signal.SIGTERM, cancel, pending)
575 except NotImplementedError:
576 # There are no good alternatives for these on Windows.
579 done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
581 src = tasks.pop(task)
583 cancelled.append(task)
584 elif task.exception():
585 report.failed(src, str(task.exception()))
587 changed = Changed.YES if task.result() else Changed.NO
588 # If the file was written back or was successfully checked as
589 # well-formatted, store this information in the cache.
590 if write_back is WriteBack.YES or (
591 write_back is WriteBack.CHECK and changed is Changed.NO
593 sources_to_cache.append(src)
594 report.done(src, changed)
596 await asyncio.gather(*cancelled, loop=loop, return_exceptions=True)
598 write_cache(cache, sources_to_cache, mode)
601 def format_file_in_place(
605 write_back: WriteBack = WriteBack.NO,
606 lock: Any = None, # multiprocessing.Manager().Lock() is some crazy proxy
608 """Format file under `src` path. Return True if changed.
610 If `write_back` is DIFF, write a diff to stdout. If it is YES, write reformatted
612 `mode` and `fast` options are passed to :func:`format_file_contents`.
614 if src.suffix == ".pyi":
615 mode = evolve(mode, is_pyi=True)
617 then = datetime.utcfromtimestamp(src.stat().st_mtime)
618 with open(src, "rb") as buf:
619 src_contents, encoding, newline = decode_bytes(buf.read())
621 dst_contents = format_file_contents(src_contents, fast=fast, mode=mode)
622 except NothingChanged:
625 if write_back == write_back.YES:
626 with open(src, "w", encoding=encoding, newline=newline) as f:
627 f.write(dst_contents)
628 elif write_back == write_back.DIFF:
629 now = datetime.utcnow()
630 src_name = f"{src}\t{then} +0000"
631 dst_name = f"{src}\t{now} +0000"
632 diff_contents = diff(src_contents, dst_contents, src_name, dst_name)
634 with lock or nullcontext():
635 f = io.TextIOWrapper(
641 f.write(diff_contents)
647 def format_stdin_to_stdout(
648 fast: bool, *, write_back: WriteBack = WriteBack.NO, mode: FileMode
650 """Format file on stdin. Return True if changed.
652 If `write_back` is YES, write reformatted code back to stdout. If it is DIFF,
653 write a diff to stdout. The `mode` argument is passed to
654 :func:`format_file_contents`.
656 then = datetime.utcnow()
657 src, encoding, newline = decode_bytes(sys.stdin.buffer.read())
660 dst = format_file_contents(src, fast=fast, mode=mode)
663 except NothingChanged:
667 f = io.TextIOWrapper(
668 sys.stdout.buffer, encoding=encoding, newline=newline, write_through=True
670 if write_back == WriteBack.YES:
672 elif write_back == WriteBack.DIFF:
673 now = datetime.utcnow()
674 src_name = f"STDIN\t{then} +0000"
675 dst_name = f"STDOUT\t{now} +0000"
676 f.write(diff(src, dst, src_name, dst_name))
680 def format_file_contents(
681 src_contents: str, *, fast: bool, mode: FileMode
683 """Reformat contents a file and return new contents.
685 If `fast` is False, additionally confirm that the reformatted code is
686 valid by calling :func:`assert_equivalent` and :func:`assert_stable` on it.
687 `mode` is passed to :func:`format_str`.
689 if src_contents.strip() == "":
692 dst_contents = format_str(src_contents, mode=mode)
693 if src_contents == dst_contents:
697 assert_equivalent(src_contents, dst_contents)
698 assert_stable(src_contents, dst_contents, mode=mode)
702 def format_str(src_contents: str, *, mode: FileMode) -> FileContent:
703 """Reformat a string and return new contents.
705 `mode` determines formatting options, such as how many characters per line are
708 src_node = lib2to3_parse(src_contents.lstrip(), mode.target_versions)
710 future_imports = get_future_imports(src_node)
711 if mode.target_versions:
712 versions = mode.target_versions
714 versions = detect_target_versions(src_node)
715 normalize_fmt_off(src_node)
716 lines = LineGenerator(
717 remove_u_prefix="unicode_literals" in future_imports
718 or supports_feature(versions, Feature.UNICODE_LITERALS),
720 normalize_strings=mode.string_normalization,
722 elt = EmptyLineTracker(is_pyi=mode.is_pyi)
725 split_line_features = {
727 for feature in {Feature.TRAILING_COMMA_IN_CALL, Feature.TRAILING_COMMA_IN_DEF}
728 if supports_feature(versions, feature)
730 for current_line in lines.visit(src_node):
731 for _ in range(after):
732 dst_contents.append(str(empty_line))
733 before, after = elt.maybe_empty_lines(current_line)
734 for _ in range(before):
735 dst_contents.append(str(empty_line))
736 for line in split_line(
737 current_line, line_length=mode.line_length, features=split_line_features
739 dst_contents.append(str(line))
740 return "".join(dst_contents)
743 def decode_bytes(src: bytes) -> Tuple[FileContent, Encoding, NewLine]:
744 """Return a tuple of (decoded_contents, encoding, newline).
746 `newline` is either CRLF or LF but `decoded_contents` is decoded with
747 universal newlines (i.e. only contains LF).
749 srcbuf = io.BytesIO(src)
750 encoding, lines = tokenize.detect_encoding(srcbuf.readline)
752 return "", encoding, "\n"
754 newline = "\r\n" if b"\r\n" == lines[0][-2:] else "\n"
756 with io.TextIOWrapper(srcbuf, encoding) as tiow:
757 return tiow.read(), encoding, newline
760 def get_grammars(target_versions: Set[TargetVersion]) -> List[Grammar]:
761 if not target_versions:
762 # No target_version specified, so try all grammars.
765 pygram.python_grammar_no_print_statement_no_exec_statement_async_keywords,
767 pygram.python_grammar_no_print_statement_no_exec_statement,
768 # Python 2.7 with future print_function import
769 pygram.python_grammar_no_print_statement,
771 pygram.python_grammar,
773 elif all(version.is_python2() for version in target_versions):
774 # Python 2-only code, so try Python 2 grammars.
776 # Python 2.7 with future print_function import
777 pygram.python_grammar_no_print_statement,
779 pygram.python_grammar,
782 # Python 3-compatible code, so only try Python 3 grammar.
784 # If we have to parse both, try to parse async as a keyword first
785 if not supports_feature(target_versions, Feature.ASYNC_IDENTIFIERS):
788 pygram.python_grammar_no_print_statement_no_exec_statement_async_keywords # noqa: B950
790 if not supports_feature(target_versions, Feature.ASYNC_KEYWORDS):
792 grammars.append(pygram.python_grammar_no_print_statement_no_exec_statement)
793 # At least one of the above branches must have been taken, because every Python
794 # version has exactly one of the two 'ASYNC_*' flags
798 def lib2to3_parse(src_txt: str, target_versions: Iterable[TargetVersion] = ()) -> Node:
799 """Given a string with source, return the lib2to3 Node."""
800 if src_txt[-1:] != "\n":
803 for grammar in get_grammars(set(target_versions)):
804 drv = driver.Driver(grammar, pytree.convert)
806 result = drv.parse_string(src_txt, True)
809 except ParseError as pe:
810 lineno, column = pe.context[1]
811 lines = src_txt.splitlines()
813 faulty_line = lines[lineno - 1]
815 faulty_line = "<line number missing in source>"
816 exc = InvalidInput(f"Cannot parse: {lineno}:{column}: {faulty_line}")
820 if isinstance(result, Leaf):
821 result = Node(syms.file_input, [result])
825 def lib2to3_unparse(node: Node) -> str:
826 """Given a lib2to3 node, return its string representation."""
834 class Visitor(Generic[T]):
835 """Basic lib2to3 visitor that yields things of type `T` on `visit()`."""
837 def visit(self, node: LN) -> Iterator[T]:
838 """Main method to visit `node` and its children.
840 It tries to find a `visit_*()` method for the given `node.type`, like
841 `visit_simple_stmt` for Node objects or `visit_INDENT` for Leaf objects.
842 If no dedicated `visit_*()` method is found, chooses `visit_default()`
845 Then yields objects of type `T` from the selected visitor.
848 name = token.tok_name[node.type]
850 name = type_repr(node.type)
851 yield from getattr(self, f"visit_{name}", self.visit_default)(node)
853 def visit_default(self, node: LN) -> Iterator[T]:
854 """Default `visit_*()` implementation. Recurses to children of `node`."""
855 if isinstance(node, Node):
856 for child in node.children:
857 yield from self.visit(child)
861 class DebugVisitor(Visitor[T]):
864 def visit_default(self, node: LN) -> Iterator[T]:
865 indent = " " * (2 * self.tree_depth)
866 if isinstance(node, Node):
867 _type = type_repr(node.type)
868 out(f"{indent}{_type}", fg="yellow")
870 for child in node.children:
871 yield from self.visit(child)
874 out(f"{indent}/{_type}", fg="yellow", bold=False)
876 _type = token.tok_name.get(node.type, str(node.type))
877 out(f"{indent}{_type}", fg="blue", nl=False)
879 # We don't have to handle prefixes for `Node` objects since
880 # that delegates to the first child anyway.
881 out(f" {node.prefix!r}", fg="green", bold=False, nl=False)
882 out(f" {node.value!r}", fg="blue", bold=False)
885 def show(cls, code: Union[str, Leaf, Node]) -> None:
886 """Pretty-print the lib2to3 AST of a given string of `code`.
888 Convenience method for debugging.
890 v: DebugVisitor[None] = DebugVisitor()
891 if isinstance(code, str):
892 code = lib2to3_parse(code)
896 WHITESPACE = {token.DEDENT, token.INDENT, token.NEWLINE}
907 STANDALONE_COMMENT = 153
908 token.tok_name[STANDALONE_COMMENT] = "STANDALONE_COMMENT"
909 LOGIC_OPERATORS = {"and", "or"}
934 STARS = {token.STAR, token.DOUBLESTAR}
937 syms.argument, # double star in arglist
938 syms.trailer, # single argument to call
940 syms.varargslist, # lambdas
942 UNPACKING_PARENTS = {
943 syms.atom, # single element of a list or set literal
947 syms.testlist_star_expr,
982 COMPREHENSION_PRIORITY = 20
984 TERNARY_PRIORITY = 16
987 COMPARATOR_PRIORITY = 10
998 token.DOUBLESLASH: 4,
1002 token.DOUBLESTAR: 2,
1008 class BracketTracker:
1009 """Keeps track of brackets on a line."""
1012 bracket_match: Dict[Tuple[Depth, NodeType], Leaf] = Factory(dict)
1013 delimiters: Dict[LeafID, Priority] = Factory(dict)
1014 previous: Optional[Leaf] = None
1015 _for_loop_depths: List[int] = Factory(list)
1016 _lambda_argument_depths: List[int] = Factory(list)
1018 def mark(self, leaf: Leaf) -> None:
1019 """Mark `leaf` with bracket-related metadata. Keep track of delimiters.
1021 All leaves receive an int `bracket_depth` field that stores how deep
1022 within brackets a given leaf is. 0 means there are no enclosing brackets
1023 that started on this line.
1025 If a leaf is itself a closing bracket, it receives an `opening_bracket`
1026 field that it forms a pair with. This is a one-directional link to
1027 avoid reference cycles.
1029 If a leaf is a delimiter (a token on which Black can split the line if
1030 needed) and it's on depth 0, its `id()` is stored in the tracker's
1033 if leaf.type == token.COMMENT:
1036 self.maybe_decrement_after_for_loop_variable(leaf)
1037 self.maybe_decrement_after_lambda_arguments(leaf)
1038 if leaf.type in CLOSING_BRACKETS:
1040 opening_bracket = self.bracket_match.pop((self.depth, leaf.type))
1041 leaf.opening_bracket = opening_bracket
1042 leaf.bracket_depth = self.depth
1044 delim = is_split_before_delimiter(leaf, self.previous)
1045 if delim and self.previous is not None:
1046 self.delimiters[id(self.previous)] = delim
1048 delim = is_split_after_delimiter(leaf, self.previous)
1050 self.delimiters[id(leaf)] = delim
1051 if leaf.type in OPENING_BRACKETS:
1052 self.bracket_match[self.depth, BRACKET[leaf.type]] = leaf
1054 self.previous = leaf
1055 self.maybe_increment_lambda_arguments(leaf)
1056 self.maybe_increment_for_loop_variable(leaf)
1058 def any_open_brackets(self) -> bool:
1059 """Return True if there is an yet unmatched open bracket on the line."""
1060 return bool(self.bracket_match)
1062 def max_delimiter_priority(self, exclude: Iterable[LeafID] = ()) -> Priority:
1063 """Return the highest priority of a delimiter found on the line.
1065 Values are consistent with what `is_split_*_delimiter()` return.
1066 Raises ValueError on no delimiters.
1068 return max(v for k, v in self.delimiters.items() if k not in exclude)
1070 def delimiter_count_with_priority(self, priority: Priority = 0) -> int:
1071 """Return the number of delimiters with the given `priority`.
1073 If no `priority` is passed, defaults to max priority on the line.
1075 if not self.delimiters:
1078 priority = priority or self.max_delimiter_priority()
1079 return sum(1 for p in self.delimiters.values() if p == priority)
1081 def maybe_increment_for_loop_variable(self, leaf: Leaf) -> bool:
1082 """In a for loop, or comprehension, the variables are often unpacks.
1084 To avoid splitting on the comma in this situation, increase the depth of
1085 tokens between `for` and `in`.
1087 if leaf.type == token.NAME and leaf.value == "for":
1089 self._for_loop_depths.append(self.depth)
1094 def maybe_decrement_after_for_loop_variable(self, leaf: Leaf) -> bool:
1095 """See `maybe_increment_for_loop_variable` above for explanation."""
1097 self._for_loop_depths
1098 and self._for_loop_depths[-1] == self.depth
1099 and leaf.type == token.NAME
1100 and leaf.value == "in"
1103 self._for_loop_depths.pop()
1108 def maybe_increment_lambda_arguments(self, leaf: Leaf) -> bool:
1109 """In a lambda expression, there might be more than one argument.
1111 To avoid splitting on the comma in this situation, increase the depth of
1112 tokens between `lambda` and `:`.
1114 if leaf.type == token.NAME and leaf.value == "lambda":
1116 self._lambda_argument_depths.append(self.depth)
1121 def maybe_decrement_after_lambda_arguments(self, leaf: Leaf) -> bool:
1122 """See `maybe_increment_lambda_arguments` above for explanation."""
1124 self._lambda_argument_depths
1125 and self._lambda_argument_depths[-1] == self.depth
1126 and leaf.type == token.COLON
1129 self._lambda_argument_depths.pop()
1134 def get_open_lsqb(self) -> Optional[Leaf]:
1135 """Return the most recent opening square bracket (if any)."""
1136 return self.bracket_match.get((self.depth - 1, token.RSQB))
1141 """Holds leaves and comments. Can be printed with `str(line)`."""
1144 leaves: List[Leaf] = Factory(list)
1145 comments: Dict[LeafID, List[Leaf]] = Factory(dict) # keys ordered like `leaves`
1146 bracket_tracker: BracketTracker = Factory(BracketTracker)
1147 inside_brackets: bool = False
1148 should_explode: bool = False
1150 def append(self, leaf: Leaf, preformatted: bool = False) -> None:
1151 """Add a new `leaf` to the end of the line.
1153 Unless `preformatted` is True, the `leaf` will receive a new consistent
1154 whitespace prefix and metadata applied by :class:`BracketTracker`.
1155 Trailing commas are maybe removed, unpacked for loop variables are
1156 demoted from being delimiters.
1158 Inline comments are put aside.
1160 has_value = leaf.type in BRACKETS or bool(leaf.value.strip())
1164 if token.COLON == leaf.type and self.is_class_paren_empty:
1165 del self.leaves[-2:]
1166 if self.leaves and not preformatted:
1167 # Note: at this point leaf.prefix should be empty except for
1168 # imports, for which we only preserve newlines.
1169 leaf.prefix += whitespace(
1170 leaf, complex_subscript=self.is_complex_subscript(leaf)
1172 if self.inside_brackets or not preformatted:
1173 self.bracket_tracker.mark(leaf)
1174 self.maybe_remove_trailing_comma(leaf)
1175 if not self.append_comment(leaf):
1176 self.leaves.append(leaf)
1178 def append_safe(self, leaf: Leaf, preformatted: bool = False) -> None:
1179 """Like :func:`append()` but disallow invalid standalone comment structure.
1181 Raises ValueError when any `leaf` is appended after a standalone comment
1182 or when a standalone comment is not the first leaf on the line.
1184 if self.bracket_tracker.depth == 0:
1186 raise ValueError("cannot append to standalone comments")
1188 if self.leaves and leaf.type == STANDALONE_COMMENT:
1190 "cannot append standalone comments to a populated line"
1193 self.append(leaf, preformatted=preformatted)
1196 def is_comment(self) -> bool:
1197 """Is this line a standalone comment?"""
1198 return len(self.leaves) == 1 and self.leaves[0].type == STANDALONE_COMMENT
1201 def is_decorator(self) -> bool:
1202 """Is this line a decorator?"""
1203 return bool(self) and self.leaves[0].type == token.AT
1206 def is_import(self) -> bool:
1207 """Is this an import line?"""
1208 return bool(self) and is_import(self.leaves[0])
1211 def is_class(self) -> bool:
1212 """Is this line a class definition?"""
1215 and self.leaves[0].type == token.NAME
1216 and self.leaves[0].value == "class"
1220 def is_stub_class(self) -> bool:
1221 """Is this line a class definition with a body consisting only of "..."?"""
1222 return self.is_class and self.leaves[-3:] == [
1223 Leaf(token.DOT, ".") for _ in range(3)
1227 def is_def(self) -> bool:
1228 """Is this a function definition? (Also returns True for async defs.)"""
1230 first_leaf = self.leaves[0]
1235 second_leaf: Optional[Leaf] = self.leaves[1]
1238 return (first_leaf.type == token.NAME and first_leaf.value == "def") or (
1239 first_leaf.type == token.ASYNC
1240 and second_leaf is not None
1241 and second_leaf.type == token.NAME
1242 and second_leaf.value == "def"
1246 def is_class_paren_empty(self) -> bool:
1247 """Is this a class with no base classes but using parentheses?
1249 Those are unnecessary and should be removed.
1253 and len(self.leaves) == 4
1255 and self.leaves[2].type == token.LPAR
1256 and self.leaves[2].value == "("
1257 and self.leaves[3].type == token.RPAR
1258 and self.leaves[3].value == ")"
1262 def is_triple_quoted_string(self) -> bool:
1263 """Is the line a triple quoted string?"""
1266 and self.leaves[0].type == token.STRING
1267 and self.leaves[0].value.startswith(('"""', "'''"))
1270 def contains_standalone_comments(self, depth_limit: int = sys.maxsize) -> bool:
1271 """If so, needs to be split before emitting."""
1272 for leaf in self.leaves:
1273 if leaf.type == STANDALONE_COMMENT:
1274 if leaf.bracket_depth <= depth_limit:
1278 def contains_inner_type_comments(self) -> bool:
1281 last_leaf = self.leaves[-1]
1282 ignored_ids.add(id(last_leaf))
1283 if last_leaf.type == token.COMMA or (
1284 last_leaf.type == token.RPAR and not last_leaf.value
1286 # When trailing commas or optional parens are inserted by Black for
1287 # consistency, comments after the previous last element are not moved
1288 # (they don't have to, rendering will still be correct). So we ignore
1289 # trailing commas and invisible.
1290 last_leaf = self.leaves[-2]
1291 ignored_ids.add(id(last_leaf))
1295 for leaf_id, comments in self.comments.items():
1296 if leaf_id in ignored_ids:
1299 for comment in comments:
1300 if is_type_comment(comment):
1305 def contains_multiline_strings(self) -> bool:
1306 for leaf in self.leaves:
1307 if is_multiline_string(leaf):
1312 def maybe_remove_trailing_comma(self, closing: Leaf) -> bool:
1313 """Remove trailing comma if there is one and it's safe."""
1316 and self.leaves[-1].type == token.COMMA
1317 and closing.type in CLOSING_BRACKETS
1321 if closing.type == token.RBRACE:
1322 self.remove_trailing_comma()
1325 if closing.type == token.RSQB:
1326 comma = self.leaves[-1]
1327 if comma.parent and comma.parent.type == syms.listmaker:
1328 self.remove_trailing_comma()
1331 # For parens let's check if it's safe to remove the comma.
1332 # Imports are always safe.
1334 self.remove_trailing_comma()
1337 # Otherwise, if the trailing one is the only one, we might mistakenly
1338 # change a tuple into a different type by removing the comma.
1339 depth = closing.bracket_depth + 1
1341 opening = closing.opening_bracket
1342 for _opening_index, leaf in enumerate(self.leaves):
1349 for leaf in self.leaves[_opening_index + 1 :]:
1353 bracket_depth = leaf.bracket_depth
1354 if bracket_depth == depth and leaf.type == token.COMMA:
1356 if leaf.parent and leaf.parent.type in {
1364 self.remove_trailing_comma()
1369 def append_comment(self, comment: Leaf) -> bool:
1370 """Add an inline or standalone comment to the line."""
1372 comment.type == STANDALONE_COMMENT
1373 and self.bracket_tracker.any_open_brackets()
1378 if comment.type != token.COMMENT:
1382 comment.type = STANDALONE_COMMENT
1386 last_leaf = self.leaves[-1]
1388 last_leaf.type == token.RPAR
1389 and not last_leaf.value
1390 and last_leaf.parent
1391 and len(list(last_leaf.parent.leaves())) <= 3
1392 and not is_type_comment(comment)
1394 # Comments on an optional parens wrapping a single leaf should belong to
1395 # the wrapped node except if it's a type comment. Pinning the comment like
1396 # this avoids unstable formatting caused by comment migration.
1397 if len(self.leaves) < 2:
1398 comment.type = STANDALONE_COMMENT
1401 last_leaf = self.leaves[-2]
1402 self.comments.setdefault(id(last_leaf), []).append(comment)
1405 def comments_after(self, leaf: Leaf) -> List[Leaf]:
1406 """Generate comments that should appear directly after `leaf`."""
1407 return self.comments.get(id(leaf), [])
1409 def remove_trailing_comma(self) -> None:
1410 """Remove the trailing comma and moves the comments attached to it."""
1411 trailing_comma = self.leaves.pop()
1412 trailing_comma_comments = self.comments.pop(id(trailing_comma), [])
1413 self.comments.setdefault(id(self.leaves[-1]), []).extend(
1414 trailing_comma_comments
1417 def is_complex_subscript(self, leaf: Leaf) -> bool:
1418 """Return True iff `leaf` is part of a slice with non-trivial exprs."""
1419 open_lsqb = self.bracket_tracker.get_open_lsqb()
1420 if open_lsqb is None:
1423 subscript_start = open_lsqb.next_sibling
1425 if isinstance(subscript_start, Node):
1426 if subscript_start.type == syms.listmaker:
1429 if subscript_start.type == syms.subscriptlist:
1430 subscript_start = child_towards(subscript_start, leaf)
1431 return subscript_start is not None and any(
1432 n.type in TEST_DESCENDANTS for n in subscript_start.pre_order()
1435 def __str__(self) -> str:
1436 """Render the line."""
1440 indent = " " * self.depth
1441 leaves = iter(self.leaves)
1442 first = next(leaves)
1443 res = f"{first.prefix}{indent}{first.value}"
1446 for comment in itertools.chain.from_iterable(self.comments.values()):
1450 def __bool__(self) -> bool:
1451 """Return True if the line has leaves or comments."""
1452 return bool(self.leaves or self.comments)
1456 class EmptyLineTracker:
1457 """Provides a stateful method that returns the number of potential extra
1458 empty lines needed before and after the currently processed line.
1460 Note: this tracker works on lines that haven't been split yet. It assumes
1461 the prefix of the first leaf consists of optional newlines. Those newlines
1462 are consumed by `maybe_empty_lines()` and included in the computation.
1465 is_pyi: bool = False
1466 previous_line: Optional[Line] = None
1467 previous_after: int = 0
1468 previous_defs: List[int] = Factory(list)
1470 def maybe_empty_lines(self, current_line: Line) -> Tuple[int, int]:
1471 """Return the number of extra empty lines before and after the `current_line`.
1473 This is for separating `def`, `async def` and `class` with extra empty
1474 lines (two on module-level).
1476 before, after = self._maybe_empty_lines(current_line)
1477 before -= self.previous_after
1478 self.previous_after = after
1479 self.previous_line = current_line
1480 return before, after
1482 def _maybe_empty_lines(self, current_line: Line) -> Tuple[int, int]:
1484 if current_line.depth == 0:
1485 max_allowed = 1 if self.is_pyi else 2
1486 if current_line.leaves:
1487 # Consume the first leaf's extra newlines.
1488 first_leaf = current_line.leaves[0]
1489 before = first_leaf.prefix.count("\n")
1490 before = min(before, max_allowed)
1491 first_leaf.prefix = ""
1494 depth = current_line.depth
1495 while self.previous_defs and self.previous_defs[-1] >= depth:
1496 self.previous_defs.pop()
1498 before = 0 if depth else 1
1500 before = 1 if depth else 2
1501 if current_line.is_decorator or current_line.is_def or current_line.is_class:
1502 return self._maybe_empty_lines_for_class_or_def(current_line, before)
1506 and self.previous_line.is_import
1507 and not current_line.is_import
1508 and depth == self.previous_line.depth
1510 return (before or 1), 0
1514 and self.previous_line.is_class
1515 and current_line.is_triple_quoted_string
1521 def _maybe_empty_lines_for_class_or_def(
1522 self, current_line: Line, before: int
1523 ) -> Tuple[int, int]:
1524 if not current_line.is_decorator:
1525 self.previous_defs.append(current_line.depth)
1526 if self.previous_line is None:
1527 # Don't insert empty lines before the first line in the file.
1530 if self.previous_line.is_decorator:
1533 if self.previous_line.depth < current_line.depth and (
1534 self.previous_line.is_class or self.previous_line.is_def
1539 self.previous_line.is_comment
1540 and self.previous_line.depth == current_line.depth
1546 if self.previous_line.depth > current_line.depth:
1548 elif current_line.is_class or self.previous_line.is_class:
1549 if current_line.is_stub_class and self.previous_line.is_stub_class:
1550 # No blank line between classes with an empty body
1554 elif current_line.is_def and not self.previous_line.is_def:
1555 # Blank line between a block of functions and a block of non-functions
1561 if current_line.depth and newlines:
1567 class LineGenerator(Visitor[Line]):
1568 """Generates reformatted Line objects. Empty lines are not emitted.
1570 Note: destroys the tree it's visiting by mutating prefixes of its leaves
1571 in ways that will no longer stringify to valid Python code on the tree.
1574 is_pyi: bool = False
1575 normalize_strings: bool = True
1576 current_line: Line = Factory(Line)
1577 remove_u_prefix: bool = False
1579 def line(self, indent: int = 0) -> Iterator[Line]:
1582 If the line is empty, only emit if it makes sense.
1583 If the line is too long, split it first and then generate.
1585 If any lines were generated, set up a new current_line.
1587 if not self.current_line:
1588 self.current_line.depth += indent
1589 return # Line is empty, don't emit. Creating a new one unnecessary.
1591 complete_line = self.current_line
1592 self.current_line = Line(depth=complete_line.depth + indent)
1595 def visit_default(self, node: LN) -> Iterator[Line]:
1596 """Default `visit_*()` implementation. Recurses to children of `node`."""
1597 if isinstance(node, Leaf):
1598 any_open_brackets = self.current_line.bracket_tracker.any_open_brackets()
1599 for comment in generate_comments(node):
1600 if any_open_brackets:
1601 # any comment within brackets is subject to splitting
1602 self.current_line.append(comment)
1603 elif comment.type == token.COMMENT:
1604 # regular trailing comment
1605 self.current_line.append(comment)
1606 yield from self.line()
1609 # regular standalone comment
1610 yield from self.line()
1612 self.current_line.append(comment)
1613 yield from self.line()
1615 normalize_prefix(node, inside_brackets=any_open_brackets)
1616 if self.normalize_strings and node.type == token.STRING:
1617 normalize_string_prefix(node, remove_u_prefix=self.remove_u_prefix)
1618 normalize_string_quotes(node)
1619 if node.type == token.NUMBER:
1620 normalize_numeric_literal(node)
1621 if node.type not in WHITESPACE:
1622 self.current_line.append(node)
1623 yield from super().visit_default(node)
1625 def visit_atom(self, node: Node) -> Iterator[Line]:
1626 # Always make parentheses invisible around a single node, because it should
1627 # not be needed (except in the case of yield, where removing the parentheses
1628 # produces a SyntaxError).
1630 len(node.children) == 3
1631 and isinstance(node.children[0], Leaf)
1632 and node.children[0].type == token.LPAR
1633 and isinstance(node.children[2], Leaf)
1634 and node.children[2].type == token.RPAR
1635 and isinstance(node.children[1], Leaf)
1637 node.children[1].type == token.NAME
1638 and node.children[1].value == "yield"
1641 node.children[0].value = ""
1642 node.children[2].value = ""
1643 yield from super().visit_default(node)
1645 def visit_factor(self, node: Node) -> Iterator[Line]:
1646 """Force parentheses between a unary op and a binary power:
1648 -2 ** 8 -> -(2 ** 8)
1650 child = node.children[1]
1651 if child.type == syms.power and len(child.children) == 3:
1652 lpar = Leaf(token.LPAR, "(")
1653 rpar = Leaf(token.RPAR, ")")
1654 index = child.remove() or 0
1655 node.insert_child(index, Node(syms.atom, [lpar, child, rpar]))
1656 yield from self.visit_default(node)
1658 def visit_INDENT(self, node: Node) -> Iterator[Line]:
1659 """Increase indentation level, maybe yield a line."""
1660 # In blib2to3 INDENT never holds comments.
1661 yield from self.line(+1)
1662 yield from self.visit_default(node)
1664 def visit_DEDENT(self, node: Node) -> Iterator[Line]:
1665 """Decrease indentation level, maybe yield a line."""
1666 # The current line might still wait for trailing comments. At DEDENT time
1667 # there won't be any (they would be prefixes on the preceding NEWLINE).
1668 # Emit the line then.
1669 yield from self.line()
1671 # While DEDENT has no value, its prefix may contain standalone comments
1672 # that belong to the current indentation level. Get 'em.
1673 yield from self.visit_default(node)
1675 # Finally, emit the dedent.
1676 yield from self.line(-1)
1679 self, node: Node, keywords: Set[str], parens: Set[str]
1680 ) -> Iterator[Line]:
1681 """Visit a statement.
1683 This implementation is shared for `if`, `while`, `for`, `try`, `except`,
1684 `def`, `with`, `class`, `assert` and assignments.
1686 The relevant Python language `keywords` for a given statement will be
1687 NAME leaves within it. This methods puts those on a separate line.
1689 `parens` holds a set of string leaf values immediately after which
1690 invisible parens should be put.
1692 normalize_invisible_parens(node, parens_after=parens)
1693 for child in node.children:
1694 if child.type == token.NAME and child.value in keywords: # type: ignore
1695 yield from self.line()
1697 yield from self.visit(child)
1699 def visit_suite(self, node: Node) -> Iterator[Line]:
1700 """Visit a suite."""
1701 if self.is_pyi and is_stub_suite(node):
1702 yield from self.visit(node.children[2])
1704 yield from self.visit_default(node)
1706 def visit_simple_stmt(self, node: Node) -> Iterator[Line]:
1707 """Visit a statement without nested statements."""
1708 is_suite_like = node.parent and node.parent.type in STATEMENT
1710 if self.is_pyi and is_stub_body(node):
1711 yield from self.visit_default(node)
1713 yield from self.line(+1)
1714 yield from self.visit_default(node)
1715 yield from self.line(-1)
1718 if not self.is_pyi or not node.parent or not is_stub_suite(node.parent):
1719 yield from self.line()
1720 yield from self.visit_default(node)
1722 def visit_async_stmt(self, node: Node) -> Iterator[Line]:
1723 """Visit `async def`, `async for`, `async with`."""
1724 yield from self.line()
1726 children = iter(node.children)
1727 for child in children:
1728 yield from self.visit(child)
1730 if child.type == token.ASYNC:
1733 internal_stmt = next(children)
1734 for child in internal_stmt.children:
1735 yield from self.visit(child)
1737 def visit_decorators(self, node: Node) -> Iterator[Line]:
1738 """Visit decorators."""
1739 for child in node.children:
1740 yield from self.line()
1741 yield from self.visit(child)
1743 def visit_SEMI(self, leaf: Leaf) -> Iterator[Line]:
1744 """Remove a semicolon and put the other statement on a separate line."""
1745 yield from self.line()
1747 def visit_ENDMARKER(self, leaf: Leaf) -> Iterator[Line]:
1748 """End of file. Process outstanding comments and end with a newline."""
1749 yield from self.visit_default(leaf)
1750 yield from self.line()
1752 def visit_STANDALONE_COMMENT(self, leaf: Leaf) -> Iterator[Line]:
1753 if not self.current_line.bracket_tracker.any_open_brackets():
1754 yield from self.line()
1755 yield from self.visit_default(leaf)
1757 def __attrs_post_init__(self) -> None:
1758 """You are in a twisty little maze of passages."""
1761 self.visit_assert_stmt = partial(v, keywords={"assert"}, parens={"assert", ","})
1762 self.visit_if_stmt = partial(
1763 v, keywords={"if", "else", "elif"}, parens={"if", "elif"}
1765 self.visit_while_stmt = partial(v, keywords={"while", "else"}, parens={"while"})
1766 self.visit_for_stmt = partial(v, keywords={"for", "else"}, parens={"for", "in"})
1767 self.visit_try_stmt = partial(
1768 v, keywords={"try", "except", "else", "finally"}, parens=Ø
1770 self.visit_except_clause = partial(v, keywords={"except"}, parens=Ø)
1771 self.visit_with_stmt = partial(v, keywords={"with"}, parens=Ø)
1772 self.visit_funcdef = partial(v, keywords={"def"}, parens=Ø)
1773 self.visit_classdef = partial(v, keywords={"class"}, parens=Ø)
1774 self.visit_expr_stmt = partial(v, keywords=Ø, parens=ASSIGNMENTS)
1775 self.visit_return_stmt = partial(v, keywords={"return"}, parens={"return"})
1776 self.visit_import_from = partial(v, keywords=Ø, parens={"import"})
1777 self.visit_del_stmt = partial(v, keywords=Ø, parens={"del"})
1778 self.visit_async_funcdef = self.visit_async_stmt
1779 self.visit_decorated = self.visit_decorators
1782 IMPLICIT_TUPLE = {syms.testlist, syms.testlist_star_expr, syms.exprlist}
1783 BRACKET = {token.LPAR: token.RPAR, token.LSQB: token.RSQB, token.LBRACE: token.RBRACE}
1784 OPENING_BRACKETS = set(BRACKET.keys())
1785 CLOSING_BRACKETS = set(BRACKET.values())
1786 BRACKETS = OPENING_BRACKETS | CLOSING_BRACKETS
1787 ALWAYS_NO_SPACE = CLOSING_BRACKETS | {token.COMMA, STANDALONE_COMMENT}
1790 def whitespace(leaf: Leaf, *, complex_subscript: bool) -> str: # noqa: C901
1791 """Return whitespace prefix if needed for the given `leaf`.
1793 `complex_subscript` signals whether the given leaf is part of a subscription
1794 which has non-trivial arguments, like arithmetic expressions or function calls.
1802 if t in ALWAYS_NO_SPACE:
1805 if t == token.COMMENT:
1808 assert p is not None, f"INTERNAL ERROR: hand-made leaf without parent: {leaf!r}"
1809 if t == token.COLON and p.type not in {
1816 prev = leaf.prev_sibling
1818 prevp = preceding_leaf(p)
1819 if not prevp or prevp.type in OPENING_BRACKETS:
1822 if t == token.COLON:
1823 if prevp.type == token.COLON:
1826 elif prevp.type != token.COMMA and not complex_subscript:
1831 if prevp.type == token.EQUAL:
1833 if prevp.parent.type in {
1841 elif prevp.parent.type == syms.typedargslist:
1842 # A bit hacky: if the equal sign has whitespace, it means we
1843 # previously found it's a typed argument. So, we're using
1847 elif prevp.type in STARS:
1848 if is_vararg(prevp, within=VARARGS_PARENTS | UNPACKING_PARENTS):
1851 elif prevp.type == token.COLON:
1852 if prevp.parent and prevp.parent.type in {syms.subscript, syms.sliceop}:
1853 return SPACE if complex_subscript else NO
1857 and prevp.parent.type == syms.factor
1858 and prevp.type in MATH_OPERATORS
1863 prevp.type == token.RIGHTSHIFT
1865 and prevp.parent.type == syms.shift_expr
1866 and prevp.prev_sibling
1867 and prevp.prev_sibling.type == token.NAME
1868 and prevp.prev_sibling.value == "print" # type: ignore
1870 # Python 2 print chevron
1873 elif prev.type in OPENING_BRACKETS:
1876 if p.type in {syms.parameters, syms.arglist}:
1877 # untyped function signatures or calls
1878 if not prev or prev.type != token.COMMA:
1881 elif p.type == syms.varargslist:
1883 if prev and prev.type != token.COMMA:
1886 elif p.type == syms.typedargslist:
1887 # typed function signatures
1891 if t == token.EQUAL:
1892 if prev.type != syms.tname:
1895 elif prev.type == token.EQUAL:
1896 # A bit hacky: if the equal sign has whitespace, it means we
1897 # previously found it's a typed argument. So, we're using that, too.
1900 elif prev.type != token.COMMA:
1903 elif p.type == syms.tname:
1906 prevp = preceding_leaf(p)
1907 if not prevp or prevp.type != token.COMMA:
1910 elif p.type == syms.trailer:
1911 # attributes and calls
1912 if t == token.LPAR or t == token.RPAR:
1917 prevp = preceding_leaf(p)
1918 if not prevp or prevp.type != token.NUMBER:
1921 elif t == token.LSQB:
1924 elif prev.type != token.COMMA:
1927 elif p.type == syms.argument:
1929 if t == token.EQUAL:
1933 prevp = preceding_leaf(p)
1934 if not prevp or prevp.type == token.LPAR:
1937 elif prev.type in {token.EQUAL} | STARS:
1940 elif p.type == syms.decorator:
1944 elif p.type == syms.dotted_name:
1948 prevp = preceding_leaf(p)
1949 if not prevp or prevp.type == token.AT or prevp.type == token.DOT:
1952 elif p.type == syms.classdef:
1956 if prev and prev.type == token.LPAR:
1959 elif p.type in {syms.subscript, syms.sliceop}:
1962 assert p.parent is not None, "subscripts are always parented"
1963 if p.parent.type == syms.subscriptlist:
1968 elif not complex_subscript:
1971 elif p.type == syms.atom:
1972 if prev and t == token.DOT:
1973 # dots, but not the first one.
1976 elif p.type == syms.dictsetmaker:
1978 if prev and prev.type == token.DOUBLESTAR:
1981 elif p.type in {syms.factor, syms.star_expr}:
1984 prevp = preceding_leaf(p)
1985 if not prevp or prevp.type in OPENING_BRACKETS:
1988 prevp_parent = prevp.parent
1989 assert prevp_parent is not None
1990 if prevp.type == token.COLON and prevp_parent.type in {
1996 elif prevp.type == token.EQUAL and prevp_parent.type == syms.argument:
1999 elif t in {token.NAME, token.NUMBER, token.STRING}:
2002 elif p.type == syms.import_from:
2004 if prev and prev.type == token.DOT:
2007 elif t == token.NAME:
2011 if prev and prev.type == token.DOT:
2014 elif p.type == syms.sliceop:
2020 def preceding_leaf(node: Optional[LN]) -> Optional[Leaf]:
2021 """Return the first leaf that precedes `node`, if any."""
2023 res = node.prev_sibling
2025 if isinstance(res, Leaf):
2029 return list(res.leaves())[-1]
2038 def child_towards(ancestor: Node, descendant: LN) -> Optional[LN]:
2039 """Return the child of `ancestor` that contains `descendant`."""
2040 node: Optional[LN] = descendant
2041 while node and node.parent != ancestor:
2046 def container_of(leaf: Leaf) -> LN:
2047 """Return `leaf` or one of its ancestors that is the topmost container of it.
2049 By "container" we mean a node where `leaf` is the very first child.
2051 same_prefix = leaf.prefix
2052 container: LN = leaf
2054 parent = container.parent
2058 if parent.children[0].prefix != same_prefix:
2061 if parent.type == syms.file_input:
2064 if parent.prev_sibling is not None and parent.prev_sibling.type in BRACKETS:
2071 def is_split_after_delimiter(leaf: Leaf, previous: Optional[Leaf] = None) -> Priority:
2072 """Return the priority of the `leaf` delimiter, given a line break after it.
2074 The delimiter priorities returned here are from those delimiters that would
2075 cause a line break after themselves.
2077 Higher numbers are higher priority.
2079 if leaf.type == token.COMMA:
2080 return COMMA_PRIORITY
2085 def is_split_before_delimiter(leaf: Leaf, previous: Optional[Leaf] = None) -> Priority:
2086 """Return the priority of the `leaf` delimiter, given a line break before it.
2088 The delimiter priorities returned here are from those delimiters that would
2089 cause a line break before themselves.
2091 Higher numbers are higher priority.
2093 if is_vararg(leaf, within=VARARGS_PARENTS | UNPACKING_PARENTS):
2094 # * and ** might also be MATH_OPERATORS but in this case they are not.
2095 # Don't treat them as a delimiter.
2099 leaf.type == token.DOT
2101 and leaf.parent.type not in {syms.import_from, syms.dotted_name}
2102 and (previous is None or previous.type in CLOSING_BRACKETS)
2107 leaf.type in MATH_OPERATORS
2109 and leaf.parent.type not in {syms.factor, syms.star_expr}
2111 return MATH_PRIORITIES[leaf.type]
2113 if leaf.type in COMPARATORS:
2114 return COMPARATOR_PRIORITY
2117 leaf.type == token.STRING
2118 and previous is not None
2119 and previous.type == token.STRING
2121 return STRING_PRIORITY
2123 if leaf.type not in {token.NAME, token.ASYNC}:
2129 and leaf.parent.type in {syms.comp_for, syms.old_comp_for}
2130 or leaf.type == token.ASYNC
2133 not isinstance(leaf.prev_sibling, Leaf)
2134 or leaf.prev_sibling.value != "async"
2136 return COMPREHENSION_PRIORITY
2141 and leaf.parent.type in {syms.comp_if, syms.old_comp_if}
2143 return COMPREHENSION_PRIORITY
2145 if leaf.value in {"if", "else"} and leaf.parent and leaf.parent.type == syms.test:
2146 return TERNARY_PRIORITY
2148 if leaf.value == "is":
2149 return COMPARATOR_PRIORITY
2154 and leaf.parent.type in {syms.comp_op, syms.comparison}
2156 previous is not None
2157 and previous.type == token.NAME
2158 and previous.value == "not"
2161 return COMPARATOR_PRIORITY
2166 and leaf.parent.type == syms.comp_op
2168 previous is not None
2169 and previous.type == token.NAME
2170 and previous.value == "is"
2173 return COMPARATOR_PRIORITY
2175 if leaf.value in LOGIC_OPERATORS and leaf.parent:
2176 return LOGIC_PRIORITY
2181 FMT_OFF = {"# fmt: off", "# fmt:off", "# yapf: disable"}
2182 FMT_ON = {"# fmt: on", "# fmt:on", "# yapf: enable"}
2185 def generate_comments(leaf: LN) -> Iterator[Leaf]:
2186 """Clean the prefix of the `leaf` and generate comments from it, if any.
2188 Comments in lib2to3 are shoved into the whitespace prefix. This happens
2189 in `pgen2/driver.py:Driver.parse_tokens()`. This was a brilliant implementation
2190 move because it does away with modifying the grammar to include all the
2191 possible places in which comments can be placed.
2193 The sad consequence for us though is that comments don't "belong" anywhere.
2194 This is why this function generates simple parentless Leaf objects for
2195 comments. We simply don't know what the correct parent should be.
2197 No matter though, we can live without this. We really only need to
2198 differentiate between inline and standalone comments. The latter don't
2199 share the line with any code.
2201 Inline comments are emitted as regular token.COMMENT leaves. Standalone
2202 are emitted with a fake STANDALONE_COMMENT token identifier.
2204 for pc in list_comments(leaf.prefix, is_endmarker=leaf.type == token.ENDMARKER):
2205 yield Leaf(pc.type, pc.value, prefix="\n" * pc.newlines)
2210 """Describes a piece of syntax that is a comment.
2212 It's not a :class:`blib2to3.pytree.Leaf` so that:
2214 * it can be cached (`Leaf` objects should not be reused more than once as
2215 they store their lineno, column, prefix, and parent information);
2216 * `newlines` and `consumed` fields are kept separate from the `value`. This
2217 simplifies handling of special marker comments like ``# fmt: off/on``.
2220 type: int # token.COMMENT or STANDALONE_COMMENT
2221 value: str # content of the comment
2222 newlines: int # how many newlines before the comment
2223 consumed: int # how many characters of the original leaf's prefix did we consume
2226 @lru_cache(maxsize=4096)
2227 def list_comments(prefix: str, *, is_endmarker: bool) -> List[ProtoComment]:
2228 """Return a list of :class:`ProtoComment` objects parsed from the given `prefix`."""
2229 result: List[ProtoComment] = []
2230 if not prefix or "#" not in prefix:
2236 for index, line in enumerate(prefix.split("\n")):
2237 consumed += len(line) + 1 # adding the length of the split '\n'
2238 line = line.lstrip()
2241 if not line.startswith("#"):
2242 # Escaped newlines outside of a comment are not really newlines at
2243 # all. We treat a single-line comment following an escaped newline
2244 # as a simple trailing comment.
2245 if line.endswith("\\"):
2249 if index == ignored_lines and not is_endmarker:
2250 comment_type = token.COMMENT # simple trailing comment
2252 comment_type = STANDALONE_COMMENT
2253 comment = make_comment(line)
2256 type=comment_type, value=comment, newlines=nlines, consumed=consumed
2263 def make_comment(content: str) -> str:
2264 """Return a consistently formatted comment from the given `content` string.
2266 All comments (except for "##", "#!", "#:", '#'", "#%%") should have a single
2267 space between the hash sign and the content.
2269 If `content` didn't start with a hash sign, one is provided.
2271 content = content.rstrip()
2275 if content[0] == "#":
2276 content = content[1:]
2277 if content and content[0] not in " !:#'%":
2278 content = " " + content
2279 return "#" + content
2285 inner: bool = False,
2286 features: Collection[Feature] = (),
2287 ) -> Iterator[Line]:
2288 """Split a `line` into potentially many lines.
2290 They should fit in the allotted `line_length` but might not be able to.
2291 `inner` signifies that there were a pair of brackets somewhere around the
2292 current `line`, possibly transitively. This means we can fallback to splitting
2293 by delimiters if the LHS/RHS don't yield any results.
2295 `features` are syntactical features that may be used in the output.
2301 line_str = str(line).strip("\n")
2304 not line.contains_inner_type_comments()
2305 and not line.should_explode
2306 and is_line_short_enough(line, line_length=line_length, line_str=line_str)
2311 split_funcs: List[SplitFunc]
2313 split_funcs = [left_hand_split]
2316 def rhs(line: Line, features: Collection[Feature]) -> Iterator[Line]:
2317 for omit in generate_trailers_to_omit(line, line_length):
2318 lines = list(right_hand_split(line, line_length, features, omit=omit))
2319 if is_line_short_enough(lines[0], line_length=line_length):
2323 # All splits failed, best effort split with no omits.
2324 # This mostly happens to multiline strings that are by definition
2325 # reported as not fitting a single line.
2326 yield from right_hand_split(line, line_length, features=features)
2328 if line.inside_brackets:
2329 split_funcs = [delimiter_split, standalone_comment_split, rhs]
2332 for split_func in split_funcs:
2333 # We are accumulating lines in `result` because we might want to abort
2334 # mission and return the original line in the end, or attempt a different
2336 result: List[Line] = []
2338 for l in split_func(line, features):
2339 if str(l).strip("\n") == line_str:
2340 raise CannotSplit("Split function returned an unchanged result")
2344 l, line_length=line_length, inner=True, features=features
2358 def left_hand_split(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2359 """Split line into many lines, starting with the first matching bracket pair.
2361 Note: this usually looks weird, only use this for function definitions.
2362 Prefer RHS otherwise. This is why this function is not symmetrical with
2363 :func:`right_hand_split` which also handles optional parentheses.
2365 tail_leaves: List[Leaf] = []
2366 body_leaves: List[Leaf] = []
2367 head_leaves: List[Leaf] = []
2368 current_leaves = head_leaves
2369 matching_bracket = None
2370 for leaf in line.leaves:
2372 current_leaves is body_leaves
2373 and leaf.type in CLOSING_BRACKETS
2374 and leaf.opening_bracket is matching_bracket
2376 current_leaves = tail_leaves if body_leaves else head_leaves
2377 current_leaves.append(leaf)
2378 if current_leaves is head_leaves:
2379 if leaf.type in OPENING_BRACKETS:
2380 matching_bracket = leaf
2381 current_leaves = body_leaves
2382 if not matching_bracket:
2383 raise CannotSplit("No brackets found")
2385 head = bracket_split_build_line(head_leaves, line, matching_bracket)
2386 body = bracket_split_build_line(body_leaves, line, matching_bracket, is_body=True)
2387 tail = bracket_split_build_line(tail_leaves, line, matching_bracket)
2388 bracket_split_succeeded_or_raise(head, body, tail)
2389 for result in (head, body, tail):
2394 def right_hand_split(
2397 features: Collection[Feature] = (),
2398 omit: Collection[LeafID] = (),
2399 ) -> Iterator[Line]:
2400 """Split line into many lines, starting with the last matching bracket pair.
2402 If the split was by optional parentheses, attempt splitting without them, too.
2403 `omit` is a collection of closing bracket IDs that shouldn't be considered for
2406 Note: running this function modifies `bracket_depth` on the leaves of `line`.
2408 tail_leaves: List[Leaf] = []
2409 body_leaves: List[Leaf] = []
2410 head_leaves: List[Leaf] = []
2411 current_leaves = tail_leaves
2412 opening_bracket = None
2413 closing_bracket = None
2414 for leaf in reversed(line.leaves):
2415 if current_leaves is body_leaves:
2416 if leaf is opening_bracket:
2417 current_leaves = head_leaves if body_leaves else tail_leaves
2418 current_leaves.append(leaf)
2419 if current_leaves is tail_leaves:
2420 if leaf.type in CLOSING_BRACKETS and id(leaf) not in omit:
2421 opening_bracket = leaf.opening_bracket
2422 closing_bracket = leaf
2423 current_leaves = body_leaves
2424 if not (opening_bracket and closing_bracket and head_leaves):
2425 # If there is no opening or closing_bracket that means the split failed and
2426 # all content is in the tail. Otherwise, if `head_leaves` are empty, it means
2427 # the matching `opening_bracket` wasn't available on `line` anymore.
2428 raise CannotSplit("No brackets found")
2430 tail_leaves.reverse()
2431 body_leaves.reverse()
2432 head_leaves.reverse()
2433 head = bracket_split_build_line(head_leaves, line, opening_bracket)
2434 body = bracket_split_build_line(body_leaves, line, opening_bracket, is_body=True)
2435 tail = bracket_split_build_line(tail_leaves, line, opening_bracket)
2436 bracket_split_succeeded_or_raise(head, body, tail)
2438 # the body shouldn't be exploded
2439 not body.should_explode
2440 # the opening bracket is an optional paren
2441 and opening_bracket.type == token.LPAR
2442 and not opening_bracket.value
2443 # the closing bracket is an optional paren
2444 and closing_bracket.type == token.RPAR
2445 and not closing_bracket.value
2446 # it's not an import (optional parens are the only thing we can split on
2447 # in this case; attempting a split without them is a waste of time)
2448 and not line.is_import
2449 # there are no standalone comments in the body
2450 and not body.contains_standalone_comments(0)
2451 # and we can actually remove the parens
2452 and can_omit_invisible_parens(body, line_length)
2454 omit = {id(closing_bracket), *omit}
2456 yield from right_hand_split(line, line_length, features=features, omit=omit)
2462 or is_line_short_enough(body, line_length=line_length)
2465 "Splitting failed, body is still too long and can't be split."
2468 elif head.contains_multiline_strings() or tail.contains_multiline_strings():
2470 "The current optional pair of parentheses is bound to fail to "
2471 "satisfy the splitting algorithm because the head or the tail "
2472 "contains multiline strings which by definition never fit one "
2476 ensure_visible(opening_bracket)
2477 ensure_visible(closing_bracket)
2478 for result in (head, body, tail):
2483 def bracket_split_succeeded_or_raise(head: Line, body: Line, tail: Line) -> None:
2484 """Raise :exc:`CannotSplit` if the last left- or right-hand split failed.
2486 Do nothing otherwise.
2488 A left- or right-hand split is based on a pair of brackets. Content before
2489 (and including) the opening bracket is left on one line, content inside the
2490 brackets is put on a separate line, and finally content starting with and
2491 following the closing bracket is put on a separate line.
2493 Those are called `head`, `body`, and `tail`, respectively. If the split
2494 produced the same line (all content in `head`) or ended up with an empty `body`
2495 and the `tail` is just the closing bracket, then it's considered failed.
2497 tail_len = len(str(tail).strip())
2500 raise CannotSplit("Splitting brackets produced the same line")
2504 f"Splitting brackets on an empty body to save "
2505 f"{tail_len} characters is not worth it"
2509 def bracket_split_build_line(
2510 leaves: List[Leaf], original: Line, opening_bracket: Leaf, *, is_body: bool = False
2512 """Return a new line with given `leaves` and respective comments from `original`.
2514 If `is_body` is True, the result line is one-indented inside brackets and as such
2515 has its first leaf's prefix normalized and a trailing comma added when expected.
2517 result = Line(depth=original.depth)
2519 result.inside_brackets = True
2522 # Since body is a new indent level, remove spurious leading whitespace.
2523 normalize_prefix(leaves[0], inside_brackets=True)
2524 # Ensure a trailing comma for imports and standalone function arguments, but
2525 # be careful not to add one after any comments.
2526 no_commas = original.is_def and not any(
2527 l.type == token.COMMA for l in leaves
2530 if original.is_import or no_commas:
2531 for i in range(len(leaves) - 1, -1, -1):
2532 if leaves[i].type == STANDALONE_COMMENT:
2534 elif leaves[i].type == token.COMMA:
2537 leaves.insert(i + 1, Leaf(token.COMMA, ","))
2541 result.append(leaf, preformatted=True)
2542 for comment_after in original.comments_after(leaf):
2543 result.append(comment_after, preformatted=True)
2545 result.should_explode = should_explode(result, opening_bracket)
2549 def dont_increase_indentation(split_func: SplitFunc) -> SplitFunc:
2550 """Normalize prefix of the first leaf in every line returned by `split_func`.
2552 This is a decorator over relevant split functions.
2556 def split_wrapper(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2557 for l in split_func(line, features):
2558 normalize_prefix(l.leaves[0], inside_brackets=True)
2561 return split_wrapper
2564 @dont_increase_indentation
2565 def delimiter_split(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2566 """Split according to delimiters of the highest priority.
2568 If the appropriate Features are given, the split will add trailing commas
2569 also in function signatures and calls that contain `*` and `**`.
2572 last_leaf = line.leaves[-1]
2574 raise CannotSplit("Line empty")
2576 bt = line.bracket_tracker
2578 delimiter_priority = bt.max_delimiter_priority(exclude={id(last_leaf)})
2580 raise CannotSplit("No delimiters found")
2582 if delimiter_priority == DOT_PRIORITY:
2583 if bt.delimiter_count_with_priority(delimiter_priority) == 1:
2584 raise CannotSplit("Splitting a single attribute from its owner looks wrong")
2586 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2587 lowest_depth = sys.maxsize
2588 trailing_comma_safe = True
2590 def append_to_line(leaf: Leaf) -> Iterator[Line]:
2591 """Append `leaf` to current line or to new line if appending impossible."""
2592 nonlocal current_line
2594 current_line.append_safe(leaf, preformatted=True)
2598 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2599 current_line.append(leaf)
2601 for leaf in line.leaves:
2602 yield from append_to_line(leaf)
2604 for comment_after in line.comments_after(leaf):
2605 yield from append_to_line(comment_after)
2607 lowest_depth = min(lowest_depth, leaf.bracket_depth)
2608 if leaf.bracket_depth == lowest_depth:
2609 if is_vararg(leaf, within={syms.typedargslist}):
2610 trailing_comma_safe = (
2611 trailing_comma_safe and Feature.TRAILING_COMMA_IN_DEF in features
2613 elif is_vararg(leaf, within={syms.arglist, syms.argument}):
2614 trailing_comma_safe = (
2615 trailing_comma_safe and Feature.TRAILING_COMMA_IN_CALL in features
2618 leaf_priority = bt.delimiters.get(id(leaf))
2619 if leaf_priority == delimiter_priority:
2622 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2626 and delimiter_priority == COMMA_PRIORITY
2627 and current_line.leaves[-1].type != token.COMMA
2628 and current_line.leaves[-1].type != STANDALONE_COMMENT
2630 current_line.append(Leaf(token.COMMA, ","))
2634 @dont_increase_indentation
2635 def standalone_comment_split(
2636 line: Line, features: Collection[Feature] = ()
2637 ) -> Iterator[Line]:
2638 """Split standalone comments from the rest of the line."""
2639 if not line.contains_standalone_comments(0):
2640 raise CannotSplit("Line does not have any standalone comments")
2642 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2644 def append_to_line(leaf: Leaf) -> Iterator[Line]:
2645 """Append `leaf` to current line or to new line if appending impossible."""
2646 nonlocal current_line
2648 current_line.append_safe(leaf, preformatted=True)
2652 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2653 current_line.append(leaf)
2655 for leaf in line.leaves:
2656 yield from append_to_line(leaf)
2658 for comment_after in line.comments_after(leaf):
2659 yield from append_to_line(comment_after)
2665 def is_import(leaf: Leaf) -> bool:
2666 """Return True if the given leaf starts an import statement."""
2673 (v == "import" and p and p.type == syms.import_name)
2674 or (v == "from" and p and p.type == syms.import_from)
2679 def is_type_comment(leaf: Leaf) -> bool:
2680 """Return True if the given leaf is a special comment.
2681 Only returns true for type comments for now."""
2684 return t in {token.COMMENT, t == STANDALONE_COMMENT} and v.startswith("# type:")
2687 def normalize_prefix(leaf: Leaf, *, inside_brackets: bool) -> None:
2688 """Leave existing extra newlines if not `inside_brackets`. Remove everything
2691 Note: don't use backslashes for formatting or you'll lose your voting rights.
2693 if not inside_brackets:
2694 spl = leaf.prefix.split("#")
2695 if "\\" not in spl[0]:
2696 nl_count = spl[-1].count("\n")
2699 leaf.prefix = "\n" * nl_count
2705 def normalize_string_prefix(leaf: Leaf, remove_u_prefix: bool = False) -> None:
2706 """Make all string prefixes lowercase.
2708 If remove_u_prefix is given, also removes any u prefix from the string.
2710 Note: Mutates its argument.
2712 match = re.match(r"^([furbFURB]*)(.*)$", leaf.value, re.DOTALL)
2713 assert match is not None, f"failed to match string {leaf.value!r}"
2714 orig_prefix = match.group(1)
2715 new_prefix = orig_prefix.lower()
2717 new_prefix = new_prefix.replace("u", "")
2718 leaf.value = f"{new_prefix}{match.group(2)}"
2721 def normalize_string_quotes(leaf: Leaf) -> None:
2722 """Prefer double quotes but only if it doesn't cause more escaping.
2724 Adds or removes backslashes as appropriate. Doesn't parse and fix
2725 strings nested in f-strings (yet).
2727 Note: Mutates its argument.
2729 value = leaf.value.lstrip("furbFURB")
2730 if value[:3] == '"""':
2733 elif value[:3] == "'''":
2736 elif value[0] == '"':
2742 first_quote_pos = leaf.value.find(orig_quote)
2743 if first_quote_pos == -1:
2744 return # There's an internal error
2746 prefix = leaf.value[:first_quote_pos]
2747 unescaped_new_quote = re.compile(rf"(([^\\]|^)(\\\\)*){new_quote}")
2748 escaped_new_quote = re.compile(rf"([^\\]|^)\\((?:\\\\)*){new_quote}")
2749 escaped_orig_quote = re.compile(rf"([^\\]|^)\\((?:\\\\)*){orig_quote}")
2750 body = leaf.value[first_quote_pos + len(orig_quote) : -len(orig_quote)]
2751 if "r" in prefix.casefold():
2752 if unescaped_new_quote.search(body):
2753 # There's at least one unescaped new_quote in this raw string
2754 # so converting is impossible
2757 # Do not introduce or remove backslashes in raw strings
2760 # remove unnecessary escapes
2761 new_body = sub_twice(escaped_new_quote, rf"\1\2{new_quote}", body)
2762 if body != new_body:
2763 # Consider the string without unnecessary escapes as the original
2765 leaf.value = f"{prefix}{orig_quote}{body}{orig_quote}"
2766 new_body = sub_twice(escaped_orig_quote, rf"\1\2{orig_quote}", new_body)
2767 new_body = sub_twice(unescaped_new_quote, rf"\1\\{new_quote}", new_body)
2768 if "f" in prefix.casefold():
2769 matches = re.findall(
2771 (?:[^{]|^)\{ # start of the string or a non-{ followed by a single {
2772 ([^{].*?) # contents of the brackets except if begins with {{
2773 \}(?:[^}]|$) # A } followed by end of the string or a non-}
2780 # Do not introduce backslashes in interpolated expressions
2782 if new_quote == '"""' and new_body[-1:] == '"':
2784 new_body = new_body[:-1] + '\\"'
2785 orig_escape_count = body.count("\\")
2786 new_escape_count = new_body.count("\\")
2787 if new_escape_count > orig_escape_count:
2788 return # Do not introduce more escaping
2790 if new_escape_count == orig_escape_count and orig_quote == '"':
2791 return # Prefer double quotes
2793 leaf.value = f"{prefix}{new_quote}{new_body}{new_quote}"
2796 def normalize_numeric_literal(leaf: Leaf) -> None:
2797 """Normalizes numeric (float, int, and complex) literals.
2799 All letters used in the representation are normalized to lowercase (except
2800 in Python 2 long literals).
2802 text = leaf.value.lower()
2803 if text.startswith(("0o", "0b")):
2804 # Leave octal and binary literals alone.
2806 elif text.startswith("0x"):
2807 # Change hex literals to upper case.
2808 before, after = text[:2], text[2:]
2809 text = f"{before}{after.upper()}"
2811 before, after = text.split("e")
2813 if after.startswith("-"):
2816 elif after.startswith("+"):
2818 before = format_float_or_int_string(before)
2819 text = f"{before}e{sign}{after}"
2820 elif text.endswith(("j", "l")):
2823 # Capitalize in "2L" because "l" looks too similar to "1".
2826 text = f"{format_float_or_int_string(number)}{suffix}"
2828 text = format_float_or_int_string(text)
2832 def format_float_or_int_string(text: str) -> str:
2833 """Formats a float string like "1.0"."""
2837 before, after = text.split(".")
2838 return f"{before or 0}.{after or 0}"
2841 def normalize_invisible_parens(node: Node, parens_after: Set[str]) -> None:
2842 """Make existing optional parentheses invisible or create new ones.
2844 `parens_after` is a set of string leaf values immediately after which parens
2847 Standardizes on visible parentheses for single-element tuples, and keeps
2848 existing visible parentheses for other tuples and generator expressions.
2850 for pc in list_comments(node.prefix, is_endmarker=False):
2851 if pc.value in FMT_OFF:
2852 # This `node` has a prefix with `# fmt: off`, don't mess with parens.
2856 for index, child in enumerate(list(node.children)):
2857 # Add parentheses around long tuple unpacking in assignments.
2860 and isinstance(child, Node)
2861 and child.type == syms.testlist_star_expr
2866 if child.type == syms.atom:
2867 if maybe_make_parens_invisible_in_atom(child, parent=node):
2868 lpar = Leaf(token.LPAR, "")
2869 rpar = Leaf(token.RPAR, "")
2870 index = child.remove() or 0
2871 node.insert_child(index, Node(syms.atom, [lpar, child, rpar]))
2872 elif is_one_tuple(child):
2873 # wrap child in visible parentheses
2874 lpar = Leaf(token.LPAR, "(")
2875 rpar = Leaf(token.RPAR, ")")
2877 node.insert_child(index, Node(syms.atom, [lpar, child, rpar]))
2878 elif node.type == syms.import_from:
2879 # "import from" nodes store parentheses directly as part of
2881 if child.type == token.LPAR:
2882 # make parentheses invisible
2883 child.value = "" # type: ignore
2884 node.children[-1].value = "" # type: ignore
2885 elif child.type != token.STAR:
2886 # insert invisible parentheses
2887 node.insert_child(index, Leaf(token.LPAR, ""))
2888 node.append_child(Leaf(token.RPAR, ""))
2891 elif not (isinstance(child, Leaf) and is_multiline_string(child)):
2892 # wrap child in invisible parentheses
2893 lpar = Leaf(token.LPAR, "")
2894 rpar = Leaf(token.RPAR, "")
2895 index = child.remove() or 0
2896 prefix = child.prefix
2898 new_child = Node(syms.atom, [lpar, child, rpar])
2899 new_child.prefix = prefix
2900 node.insert_child(index, new_child)
2902 check_lpar = isinstance(child, Leaf) and child.value in parens_after
2905 def normalize_fmt_off(node: Node) -> None:
2906 """Convert content between `# fmt: off`/`# fmt: on` into standalone comments."""
2909 try_again = convert_one_fmt_off_pair(node)
2912 def convert_one_fmt_off_pair(node: Node) -> bool:
2913 """Convert content of a single `# fmt: off`/`# fmt: on` into a standalone comment.
2915 Returns True if a pair was converted.
2917 for leaf in node.leaves():
2918 previous_consumed = 0
2919 for comment in list_comments(leaf.prefix, is_endmarker=False):
2920 if comment.value in FMT_OFF:
2921 # We only want standalone comments. If there's no previous leaf or
2922 # the previous leaf is indentation, it's a standalone comment in
2924 if comment.type != STANDALONE_COMMENT:
2925 prev = preceding_leaf(leaf)
2926 if prev and prev.type not in WHITESPACE:
2929 ignored_nodes = list(generate_ignored_nodes(leaf))
2930 if not ignored_nodes:
2933 first = ignored_nodes[0] # Can be a container node with the `leaf`.
2934 parent = first.parent
2935 prefix = first.prefix
2936 first.prefix = prefix[comment.consumed :]
2938 comment.value + "\n" + "".join(str(n) for n in ignored_nodes)
2940 if hidden_value.endswith("\n"):
2941 # That happens when one of the `ignored_nodes` ended with a NEWLINE
2942 # leaf (possibly followed by a DEDENT).
2943 hidden_value = hidden_value[:-1]
2945 for ignored in ignored_nodes:
2946 index = ignored.remove()
2947 if first_idx is None:
2949 assert parent is not None, "INTERNAL ERROR: fmt: on/off handling (1)"
2950 assert first_idx is not None, "INTERNAL ERROR: fmt: on/off handling (2)"
2951 parent.insert_child(
2956 prefix=prefix[:previous_consumed] + "\n" * comment.newlines,
2961 previous_consumed = comment.consumed
2966 def generate_ignored_nodes(leaf: Leaf) -> Iterator[LN]:
2967 """Starting from the container of `leaf`, generate all leaves until `# fmt: on`.
2969 Stops at the end of the block.
2971 container: Optional[LN] = container_of(leaf)
2972 while container is not None and container.type != token.ENDMARKER:
2973 for comment in list_comments(container.prefix, is_endmarker=False):
2974 if comment.value in FMT_ON:
2979 container = container.next_sibling
2982 def maybe_make_parens_invisible_in_atom(node: LN, parent: LN) -> bool:
2983 """If it's safe, make the parens in the atom `node` invisible, recursively.
2985 Returns whether the node should itself be wrapped in invisible parentheses.
2989 node.type != syms.atom
2990 or is_empty_tuple(node)
2991 or is_one_tuple(node)
2992 or (is_yield(node) and parent.type != syms.expr_stmt)
2993 or max_delimiter_priority_in_atom(node) >= COMMA_PRIORITY
2997 first = node.children[0]
2998 last = node.children[-1]
2999 if first.type == token.LPAR and last.type == token.RPAR:
3000 # make parentheses invisible
3001 first.value = "" # type: ignore
3002 last.value = "" # type: ignore
3003 if len(node.children) > 1:
3004 maybe_make_parens_invisible_in_atom(node.children[1], parent=parent)
3010 def is_empty_tuple(node: LN) -> bool:
3011 """Return True if `node` holds an empty tuple."""
3013 node.type == syms.atom
3014 and len(node.children) == 2
3015 and node.children[0].type == token.LPAR
3016 and node.children[1].type == token.RPAR
3020 def is_one_tuple(node: LN) -> bool:
3021 """Return True if `node` holds a tuple with one element, with or without parens."""
3022 if node.type == syms.atom:
3023 if len(node.children) != 3:
3026 lpar, gexp, rpar = node.children
3028 lpar.type == token.LPAR
3029 and gexp.type == syms.testlist_gexp
3030 and rpar.type == token.RPAR
3034 return len(gexp.children) == 2 and gexp.children[1].type == token.COMMA
3037 node.type in IMPLICIT_TUPLE
3038 and len(node.children) == 2
3039 and node.children[1].type == token.COMMA
3043 def is_yield(node: LN) -> bool:
3044 """Return True if `node` holds a `yield` or `yield from` expression."""
3045 if node.type == syms.yield_expr:
3048 if node.type == token.NAME and node.value == "yield": # type: ignore
3051 if node.type != syms.atom:
3054 if len(node.children) != 3:
3057 lpar, expr, rpar = node.children
3058 if lpar.type == token.LPAR and rpar.type == token.RPAR:
3059 return is_yield(expr)
3064 def is_vararg(leaf: Leaf, within: Set[NodeType]) -> bool:
3065 """Return True if `leaf` is a star or double star in a vararg or kwarg.
3067 If `within` includes VARARGS_PARENTS, this applies to function signatures.
3068 If `within` includes UNPACKING_PARENTS, it applies to right hand-side
3069 extended iterable unpacking (PEP 3132) and additional unpacking
3070 generalizations (PEP 448).
3072 if leaf.type not in STARS or not leaf.parent:
3076 if p.type == syms.star_expr:
3077 # Star expressions are also used as assignment targets in extended
3078 # iterable unpacking (PEP 3132). See what its parent is instead.
3084 return p.type in within
3087 def is_multiline_string(leaf: Leaf) -> bool:
3088 """Return True if `leaf` is a multiline string that actually spans many lines."""
3089 value = leaf.value.lstrip("furbFURB")
3090 return value[:3] in {'"""', "'''"} and "\n" in value
3093 def is_stub_suite(node: Node) -> bool:
3094 """Return True if `node` is a suite with a stub body."""
3096 len(node.children) != 4
3097 or node.children[0].type != token.NEWLINE
3098 or node.children[1].type != token.INDENT
3099 or node.children[3].type != token.DEDENT
3103 return is_stub_body(node.children[2])
3106 def is_stub_body(node: LN) -> bool:
3107 """Return True if `node` is a simple statement containing an ellipsis."""
3108 if not isinstance(node, Node) or node.type != syms.simple_stmt:
3111 if len(node.children) != 2:
3114 child = node.children[0]
3116 child.type == syms.atom
3117 and len(child.children) == 3
3118 and all(leaf == Leaf(token.DOT, ".") for leaf in child.children)
3122 def max_delimiter_priority_in_atom(node: LN) -> Priority:
3123 """Return maximum delimiter priority inside `node`.
3125 This is specific to atoms with contents contained in a pair of parentheses.
3126 If `node` isn't an atom or there are no enclosing parentheses, returns 0.
3128 if node.type != syms.atom:
3131 first = node.children[0]
3132 last = node.children[-1]
3133 if not (first.type == token.LPAR and last.type == token.RPAR):
3136 bt = BracketTracker()
3137 for c in node.children[1:-1]:
3138 if isinstance(c, Leaf):
3141 for leaf in c.leaves():
3144 return bt.max_delimiter_priority()
3150 def ensure_visible(leaf: Leaf) -> None:
3151 """Make sure parentheses are visible.
3153 They could be invisible as part of some statements (see
3154 :func:`normalize_invisible_parens` and :func:`visit_import_from`).
3156 if leaf.type == token.LPAR:
3158 elif leaf.type == token.RPAR:
3162 def should_explode(line: Line, opening_bracket: Leaf) -> bool:
3163 """Should `line` immediately be split with `delimiter_split()` after RHS?"""
3166 opening_bracket.parent
3167 and opening_bracket.parent.type in {syms.atom, syms.import_from}
3168 and opening_bracket.value in "[{("
3173 last_leaf = line.leaves[-1]
3174 exclude = {id(last_leaf)} if last_leaf.type == token.COMMA else set()
3175 max_priority = line.bracket_tracker.max_delimiter_priority(exclude=exclude)
3176 except (IndexError, ValueError):
3179 return max_priority == COMMA_PRIORITY
3182 def get_features_used(node: Node) -> Set[Feature]:
3183 """Return a set of (relatively) new Python features used in this file.
3185 Currently looking for:
3187 - underscores in numeric literals; and
3188 - trailing commas after * or ** in function signatures and calls.
3190 features: Set[Feature] = set()
3191 for n in node.pre_order():
3192 if n.type == token.STRING:
3193 value_head = n.value[:2] # type: ignore
3194 if value_head in {'f"', 'F"', "f'", "F'", "rf", "fr", "RF", "FR"}:
3195 features.add(Feature.F_STRINGS)
3197 elif n.type == token.NUMBER:
3198 if "_" in n.value: # type: ignore
3199 features.add(Feature.NUMERIC_UNDERSCORES)
3202 n.type in {syms.typedargslist, syms.arglist}
3204 and n.children[-1].type == token.COMMA
3206 if n.type == syms.typedargslist:
3207 feature = Feature.TRAILING_COMMA_IN_DEF
3209 feature = Feature.TRAILING_COMMA_IN_CALL
3211 for ch in n.children:
3212 if ch.type in STARS:
3213 features.add(feature)
3215 if ch.type == syms.argument:
3216 for argch in ch.children:
3217 if argch.type in STARS:
3218 features.add(feature)
3223 def detect_target_versions(node: Node) -> Set[TargetVersion]:
3224 """Detect the version to target based on the nodes used."""
3225 features = get_features_used(node)
3227 version for version in TargetVersion if features <= VERSION_TO_FEATURES[version]
3231 def generate_trailers_to_omit(line: Line, line_length: int) -> Iterator[Set[LeafID]]:
3232 """Generate sets of closing bracket IDs that should be omitted in a RHS.
3234 Brackets can be omitted if the entire trailer up to and including
3235 a preceding closing bracket fits in one line.
3237 Yielded sets are cumulative (contain results of previous yields, too). First
3241 omit: Set[LeafID] = set()
3244 length = 4 * line.depth
3245 opening_bracket = None
3246 closing_bracket = None
3247 inner_brackets: Set[LeafID] = set()
3248 for index, leaf, leaf_length in enumerate_with_length(line, reversed=True):
3249 length += leaf_length
3250 if length > line_length:
3253 has_inline_comment = leaf_length > len(leaf.value) + len(leaf.prefix)
3254 if leaf.type == STANDALONE_COMMENT or has_inline_comment:
3258 if leaf is opening_bracket:
3259 opening_bracket = None
3260 elif leaf.type in CLOSING_BRACKETS:
3261 inner_brackets.add(id(leaf))
3262 elif leaf.type in CLOSING_BRACKETS:
3263 if index > 0 and line.leaves[index - 1].type in OPENING_BRACKETS:
3264 # Empty brackets would fail a split so treat them as "inner"
3265 # brackets (e.g. only add them to the `omit` set if another
3266 # pair of brackets was good enough.
3267 inner_brackets.add(id(leaf))
3271 omit.add(id(closing_bracket))
3272 omit.update(inner_brackets)
3273 inner_brackets.clear()
3277 opening_bracket = leaf.opening_bracket
3278 closing_bracket = leaf
3281 def get_future_imports(node: Node) -> Set[str]:
3282 """Return a set of __future__ imports in the file."""
3283 imports: Set[str] = set()
3285 def get_imports_from_children(children: List[LN]) -> Generator[str, None, None]:
3286 for child in children:
3287 if isinstance(child, Leaf):
3288 if child.type == token.NAME:
3290 elif child.type == syms.import_as_name:
3291 orig_name = child.children[0]
3292 assert isinstance(orig_name, Leaf), "Invalid syntax parsing imports"
3293 assert orig_name.type == token.NAME, "Invalid syntax parsing imports"
3294 yield orig_name.value
3295 elif child.type == syms.import_as_names:
3296 yield from get_imports_from_children(child.children)
3298 raise AssertionError("Invalid syntax parsing imports")
3300 for child in node.children:
3301 if child.type != syms.simple_stmt:
3303 first_child = child.children[0]
3304 if isinstance(first_child, Leaf):
3305 # Continue looking if we see a docstring; otherwise stop.
3307 len(child.children) == 2
3308 and first_child.type == token.STRING
3309 and child.children[1].type == token.NEWLINE
3314 elif first_child.type == syms.import_from:
3315 module_name = first_child.children[1]
3316 if not isinstance(module_name, Leaf) or module_name.value != "__future__":
3318 imports |= set(get_imports_from_children(first_child.children[3:]))
3324 def gen_python_files_in_dir(
3327 include: Pattern[str],
3328 exclude: Pattern[str],
3330 ) -> Iterator[Path]:
3331 """Generate all files under `path` whose paths are not excluded by the
3332 `exclude` regex, but are included by the `include` regex.
3334 Symbolic links pointing outside of the `root` directory are ignored.
3336 `report` is where output about exclusions goes.
3338 assert root.is_absolute(), f"INTERNAL ERROR: `root` must be absolute but is {root}"
3339 for child in path.iterdir():
3341 normalized_path = "/" + child.resolve().relative_to(root).as_posix()
3343 if child.is_symlink():
3344 report.path_ignored(
3345 child, f"is a symbolic link that points outside {root}"
3352 normalized_path += "/"
3353 exclude_match = exclude.search(normalized_path)
3354 if exclude_match and exclude_match.group(0):
3355 report.path_ignored(child, f"matches the --exclude regular expression")
3359 yield from gen_python_files_in_dir(child, root, include, exclude, report)
3361 elif child.is_file():
3362 include_match = include.search(normalized_path)
3368 def find_project_root(srcs: Iterable[str]) -> Path:
3369 """Return a directory containing .git, .hg, or pyproject.toml.
3371 That directory can be one of the directories passed in `srcs` or their
3374 If no directory in the tree contains a marker that would specify it's the
3375 project root, the root of the file system is returned.
3378 return Path("/").resolve()
3380 common_base = min(Path(src).resolve() for src in srcs)
3381 if common_base.is_dir():
3382 # Append a fake file so `parents` below returns `common_base_dir`, too.
3383 common_base /= "fake-file"
3384 for directory in common_base.parents:
3385 if (directory / ".git").is_dir():
3388 if (directory / ".hg").is_dir():
3391 if (directory / "pyproject.toml").is_file():
3399 """Provides a reformatting counter. Can be rendered with `str(report)`."""
3403 verbose: bool = False
3404 change_count: int = 0
3406 failure_count: int = 0
3408 def done(self, src: Path, changed: Changed) -> None:
3409 """Increment the counter for successful reformatting. Write out a message."""
3410 if changed is Changed.YES:
3411 reformatted = "would reformat" if self.check else "reformatted"
3412 if self.verbose or not self.quiet:
3413 out(f"{reformatted} {src}")
3414 self.change_count += 1
3417 if changed is Changed.NO:
3418 msg = f"{src} already well formatted, good job."
3420 msg = f"{src} wasn't modified on disk since last run."
3421 out(msg, bold=False)
3422 self.same_count += 1
3424 def failed(self, src: Path, message: str) -> None:
3425 """Increment the counter for failed reformatting. Write out a message."""
3426 err(f"error: cannot format {src}: {message}")
3427 self.failure_count += 1
3429 def path_ignored(self, path: Path, message: str) -> None:
3431 out(f"{path} ignored: {message}", bold=False)
3434 def return_code(self) -> int:
3435 """Return the exit code that the app should use.
3437 This considers the current state of changed files and failures:
3438 - if there were any failures, return 123;
3439 - if any files were changed and --check is being used, return 1;
3440 - otherwise return 0.
3442 # According to http://tldp.org/LDP/abs/html/exitcodes.html starting with
3443 # 126 we have special return codes reserved by the shell.
3444 if self.failure_count:
3447 elif self.change_count and self.check:
3452 def __str__(self) -> str:
3453 """Render a color report of the current state.
3455 Use `click.unstyle` to remove colors.
3458 reformatted = "would be reformatted"
3459 unchanged = "would be left unchanged"
3460 failed = "would fail to reformat"
3462 reformatted = "reformatted"
3463 unchanged = "left unchanged"
3464 failed = "failed to reformat"
3466 if self.change_count:
3467 s = "s" if self.change_count > 1 else ""
3469 click.style(f"{self.change_count} file{s} {reformatted}", bold=True)
3472 s = "s" if self.same_count > 1 else ""
3473 report.append(f"{self.same_count} file{s} {unchanged}")
3474 if self.failure_count:
3475 s = "s" if self.failure_count > 1 else ""
3477 click.style(f"{self.failure_count} file{s} {failed}", fg="red")
3479 return ", ".join(report) + "."
3482 def parse_ast(src: str) -> Union[ast3.AST, ast27.AST]:
3483 for feature_version in (7, 6):
3485 return ast3.parse(src, feature_version=feature_version)
3489 return ast27.parse(src)
3492 def assert_equivalent(src: str, dst: str) -> None:
3493 """Raise AssertionError if `src` and `dst` aren't equivalent."""
3495 def _v(node: Union[ast3.AST, ast27.AST], depth: int = 0) -> Iterator[str]:
3496 """Simple visitor generating strings to compare ASTs by content."""
3497 yield f"{' ' * depth}{node.__class__.__name__}("
3499 for field in sorted(node._fields):
3500 # TypeIgnore has only one field 'lineno' which breaks this comparison
3501 if isinstance(node, (ast3.TypeIgnore, ast27.TypeIgnore)):
3504 # Ignore str kind which is case sensitive / and ignores unicode_literals
3505 if isinstance(node, (ast3.Str, ast27.Str, ast3.Bytes)) and field == "kind":
3509 value = getattr(node, field)
3510 except AttributeError:
3513 yield f"{' ' * (depth+1)}{field}="
3515 if isinstance(value, list):
3517 # Ignore nested tuples within del statements, because we may insert
3518 # parentheses and they change the AST.
3521 and isinstance(node, (ast3.Delete, ast27.Delete))
3522 and isinstance(item, (ast3.Tuple, ast27.Tuple))
3524 for item in item.elts:
3525 yield from _v(item, depth + 2)
3526 elif isinstance(item, (ast3.AST, ast27.AST)):
3527 yield from _v(item, depth + 2)
3529 elif isinstance(value, (ast3.AST, ast27.AST)):
3530 yield from _v(value, depth + 2)
3533 yield f"{' ' * (depth+2)}{value!r}, # {value.__class__.__name__}"
3535 yield f"{' ' * depth}) # /{node.__class__.__name__}"
3538 src_ast = parse_ast(src)
3539 except Exception as exc:
3540 raise AssertionError(
3541 f"cannot use --safe with this file; failed to parse source file. "
3542 f"AST error message: {exc}"
3546 dst_ast = parse_ast(dst)
3547 except Exception as exc:
3548 log = dump_to_file("".join(traceback.format_tb(exc.__traceback__)), dst)
3549 raise AssertionError(
3550 f"INTERNAL ERROR: Black produced invalid code: {exc}. "
3551 f"Please report a bug on https://github.com/python/black/issues. "
3552 f"This invalid output might be helpful: {log}"
3555 src_ast_str = "\n".join(_v(src_ast))
3556 dst_ast_str = "\n".join(_v(dst_ast))
3557 if src_ast_str != dst_ast_str:
3558 log = dump_to_file(diff(src_ast_str, dst_ast_str, "src", "dst"))
3559 raise AssertionError(
3560 f"INTERNAL ERROR: Black produced code that is not equivalent to "
3562 f"Please report a bug on https://github.com/python/black/issues. "
3563 f"This diff might be helpful: {log}"
3567 def assert_stable(src: str, dst: str, mode: FileMode) -> None:
3568 """Raise AssertionError if `dst` reformats differently the second time."""
3569 newdst = format_str(dst, mode=mode)
3572 diff(src, dst, "source", "first pass"),
3573 diff(dst, newdst, "first pass", "second pass"),
3575 raise AssertionError(
3576 f"INTERNAL ERROR: Black produced different code on the second pass "
3577 f"of the formatter. "
3578 f"Please report a bug on https://github.com/python/black/issues. "
3579 f"This diff might be helpful: {log}"
3583 def dump_to_file(*output: str) -> str:
3584 """Dump `output` to a temporary file. Return path to the file."""
3585 with tempfile.NamedTemporaryFile(
3586 mode="w", prefix="blk_", suffix=".log", delete=False, encoding="utf8"
3588 for lines in output:
3590 if lines and lines[-1] != "\n":
3596 def nullcontext() -> Iterator[None]:
3597 """Return context manager that does nothing.
3598 Similar to `nullcontext` from python 3.7"""
3602 def diff(a: str, b: str, a_name: str, b_name: str) -> str:
3603 """Return a unified diff string between strings `a` and `b`."""
3606 a_lines = [line + "\n" for line in a.split("\n")]
3607 b_lines = [line + "\n" for line in b.split("\n")]
3609 difflib.unified_diff(a_lines, b_lines, fromfile=a_name, tofile=b_name, n=5)
3613 def cancel(tasks: Iterable[asyncio.Task]) -> None:
3614 """asyncio signal handler that cancels all `tasks` and reports to stderr."""
3620 def shutdown(loop: asyncio.AbstractEventLoop) -> None:
3621 """Cancel all pending tasks on `loop`, wait for them, and close the loop."""
3623 if sys.version_info[:2] >= (3, 7):
3624 all_tasks = asyncio.all_tasks
3626 all_tasks = asyncio.Task.all_tasks
3627 # This part is borrowed from asyncio/runners.py in Python 3.7b2.
3628 to_cancel = [task for task in all_tasks(loop) if not task.done()]
3632 for task in to_cancel:
3634 loop.run_until_complete(
3635 asyncio.gather(*to_cancel, loop=loop, return_exceptions=True)
3638 # `concurrent.futures.Future` objects cannot be cancelled once they
3639 # are already running. There might be some when the `shutdown()` happened.
3640 # Silence their logger's spew about the event loop being closed.
3641 cf_logger = logging.getLogger("concurrent.futures")
3642 cf_logger.setLevel(logging.CRITICAL)
3646 def sub_twice(regex: Pattern[str], replacement: str, original: str) -> str:
3647 """Replace `regex` with `replacement` twice on `original`.
3649 This is used by string normalization to perform replaces on
3650 overlapping matches.
3652 return regex.sub(replacement, regex.sub(replacement, original))
3655 def re_compile_maybe_verbose(regex: str) -> Pattern[str]:
3656 """Compile a regular expression string in `regex`.
3658 If it contains newlines, use verbose mode.
3661 regex = "(?x)" + regex
3662 return re.compile(regex)
3665 def enumerate_reversed(sequence: Sequence[T]) -> Iterator[Tuple[Index, T]]:
3666 """Like `reversed(enumerate(sequence))` if that were possible."""
3667 index = len(sequence) - 1
3668 for element in reversed(sequence):
3669 yield (index, element)
3673 def enumerate_with_length(
3674 line: Line, reversed: bool = False
3675 ) -> Iterator[Tuple[Index, Leaf, int]]:
3676 """Return an enumeration of leaves with their length.
3678 Stops prematurely on multiline strings and standalone comments.
3681 Callable[[Sequence[Leaf]], Iterator[Tuple[Index, Leaf]]],
3682 enumerate_reversed if reversed else enumerate,
3684 for index, leaf in op(line.leaves):
3685 length = len(leaf.prefix) + len(leaf.value)
3686 if "\n" in leaf.value:
3687 return # Multiline strings, we can't continue.
3689 for comment in line.comments_after(leaf):
3690 length += len(comment.value)
3692 yield index, leaf, length
3695 def is_line_short_enough(line: Line, *, line_length: int, line_str: str = "") -> bool:
3696 """Return True if `line` is no longer than `line_length`.
3698 Uses the provided `line_str` rendering, if any, otherwise computes a new one.
3701 line_str = str(line).strip("\n")
3703 len(line_str) <= line_length
3704 and "\n" not in line_str # multiline strings
3705 and not line.contains_standalone_comments()
3709 def can_be_split(line: Line) -> bool:
3710 """Return False if the line cannot be split *for sure*.
3712 This is not an exhaustive search but a cheap heuristic that we can use to
3713 avoid some unfortunate formattings (mostly around wrapping unsplittable code
3714 in unnecessary parentheses).
3716 leaves = line.leaves
3720 if leaves[0].type == token.STRING and leaves[1].type == token.DOT:
3724 for leaf in leaves[-2::-1]:
3725 if leaf.type in OPENING_BRACKETS:
3726 if next.type not in CLOSING_BRACKETS:
3730 elif leaf.type == token.DOT:
3732 elif leaf.type == token.NAME:
3733 if not (next.type == token.DOT or next.type in OPENING_BRACKETS):
3736 elif leaf.type not in CLOSING_BRACKETS:
3739 if dot_count > 1 and call_count > 1:
3745 def can_omit_invisible_parens(line: Line, line_length: int) -> bool:
3746 """Does `line` have a shape safe to reformat without optional parens around it?
3748 Returns True for only a subset of potentially nice looking formattings but
3749 the point is to not return false positives that end up producing lines that
3752 bt = line.bracket_tracker
3753 if not bt.delimiters:
3754 # Without delimiters the optional parentheses are useless.
3757 max_priority = bt.max_delimiter_priority()
3758 if bt.delimiter_count_with_priority(max_priority) > 1:
3759 # With more than one delimiter of a kind the optional parentheses read better.
3762 if max_priority == DOT_PRIORITY:
3763 # A single stranded method call doesn't require optional parentheses.
3766 assert len(line.leaves) >= 2, "Stranded delimiter"
3768 first = line.leaves[0]
3769 second = line.leaves[1]
3770 penultimate = line.leaves[-2]
3771 last = line.leaves[-1]
3773 # With a single delimiter, omit if the expression starts or ends with
3775 if first.type in OPENING_BRACKETS and second.type not in CLOSING_BRACKETS:
3777 length = 4 * line.depth
3778 for _index, leaf, leaf_length in enumerate_with_length(line):
3779 if leaf.type in CLOSING_BRACKETS and leaf.opening_bracket is first:
3782 length += leaf_length
3783 if length > line_length:
3786 if leaf.type in OPENING_BRACKETS:
3787 # There are brackets we can further split on.
3791 # checked the entire string and line length wasn't exceeded
3792 if len(line.leaves) == _index + 1:
3795 # Note: we are not returning False here because a line might have *both*
3796 # a leading opening bracket and a trailing closing bracket. If the
3797 # opening bracket doesn't match our rule, maybe the closing will.
3800 last.type == token.RPAR
3801 or last.type == token.RBRACE
3803 # don't use indexing for omitting optional parentheses;
3805 last.type == token.RSQB
3807 and last.parent.type != syms.trailer
3810 if penultimate.type in OPENING_BRACKETS:
3811 # Empty brackets don't help.
3814 if is_multiline_string(first):
3815 # Additional wrapping of a multiline string in this situation is
3819 length = 4 * line.depth
3820 seen_other_brackets = False
3821 for _index, leaf, leaf_length in enumerate_with_length(line):
3822 length += leaf_length
3823 if leaf is last.opening_bracket:
3824 if seen_other_brackets or length <= line_length:
3827 elif leaf.type in OPENING_BRACKETS:
3828 # There are brackets we can further split on.
3829 seen_other_brackets = True
3834 def get_cache_file(mode: FileMode) -> Path:
3835 return CACHE_DIR / f"cache.{mode.get_cache_key()}.pickle"
3838 def read_cache(mode: FileMode) -> Cache:
3839 """Read the cache if it exists and is well formed.
3841 If it is not well formed, the call to write_cache later should resolve the issue.
3843 cache_file = get_cache_file(mode)
3844 if not cache_file.exists():
3847 with cache_file.open("rb") as fobj:
3849 cache: Cache = pickle.load(fobj)
3850 except pickle.UnpicklingError:
3856 def get_cache_info(path: Path) -> CacheInfo:
3857 """Return the information used to check if a file is already formatted or not."""
3859 return stat.st_mtime, stat.st_size
3862 def filter_cached(cache: Cache, sources: Iterable[Path]) -> Tuple[Set[Path], Set[Path]]:
3863 """Split an iterable of paths in `sources` into two sets.
3865 The first contains paths of files that modified on disk or are not in the
3866 cache. The other contains paths to non-modified files.
3868 todo, done = set(), set()
3871 if cache.get(src) != get_cache_info(src):
3878 def write_cache(cache: Cache, sources: Iterable[Path], mode: FileMode) -> None:
3879 """Update the cache file."""
3880 cache_file = get_cache_file(mode)
3882 CACHE_DIR.mkdir(parents=True, exist_ok=True)
3883 new_cache = {**cache, **{src.resolve(): get_cache_info(src) for src in sources}}
3884 with tempfile.NamedTemporaryFile(dir=str(cache_file.parent), delete=False) as f:
3885 pickle.dump(new_cache, f, protocol=pickle.HIGHEST_PROTOCOL)
3886 os.replace(f.name, cache_file)
3891 def patch_click() -> None:
3892 """Make Click not crash.
3894 On certain misconfigured environments, Python 3 selects the ASCII encoding as the
3895 default which restricts paths that it can access during the lifetime of the
3896 application. Click refuses to work in this scenario by raising a RuntimeError.
3898 In case of Black the likelihood that non-ASCII characters are going to be used in
3899 file paths is minimal since it's Python source code. Moreover, this crash was
3900 spurious on Python 3.7 thanks to PEP 538 and PEP 540.
3903 from click import core
3904 from click import _unicodefun # type: ignore
3905 except ModuleNotFoundError:
3908 for module in (core, _unicodefun):
3909 if hasattr(module, "_verify_python3_env"):
3910 module._verify_python3_env = lambda: None
3913 def patched_main() -> None:
3919 if __name__ == "__main__":