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.
3 from concurrent.futures import Executor, ProcessPoolExecutor
4 from contextlib import contextmanager
5 from datetime import datetime
7 from functools import lru_cache, partial, wraps
11 from multiprocessing import Manager, freeze_support
13 from pathlib import Path
41 from appdirs import user_cache_dir
42 from attr import dataclass, evolve, Factory
45 from typed_ast import ast3, ast27
46 from pathspec import PathSpec
49 from blib2to3.pytree import Node, Leaf, type_repr
50 from blib2to3 import pygram, pytree
51 from blib2to3.pgen2 import driver, token
52 from blib2to3.pgen2.grammar import Grammar
53 from blib2to3.pgen2.parse import ParseError
55 from _black_version import version as __version__
57 DEFAULT_LINE_LENGTH = 88
58 DEFAULT_EXCLUDES = r"/(\.eggs|\.git|\.hg|\.mypy_cache|\.nox|\.tox|\.venv|\.svn|_build|buck-out|build|dist)/" # noqa: B950
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
144 ASSIGNMENT_EXPRESSIONS = 8
145 POS_ONLY_ARGUMENTS = 9
148 VERSION_TO_FEATURES: Dict[TargetVersion, Set[Feature]] = {
149 TargetVersion.PY27: {Feature.ASYNC_IDENTIFIERS},
150 TargetVersion.PY33: {Feature.UNICODE_LITERALS, Feature.ASYNC_IDENTIFIERS},
151 TargetVersion.PY34: {Feature.UNICODE_LITERALS, Feature.ASYNC_IDENTIFIERS},
152 TargetVersion.PY35: {
153 Feature.UNICODE_LITERALS,
154 Feature.TRAILING_COMMA_IN_CALL,
155 Feature.ASYNC_IDENTIFIERS,
157 TargetVersion.PY36: {
158 Feature.UNICODE_LITERALS,
160 Feature.NUMERIC_UNDERSCORES,
161 Feature.TRAILING_COMMA_IN_CALL,
162 Feature.TRAILING_COMMA_IN_DEF,
163 Feature.ASYNC_IDENTIFIERS,
165 TargetVersion.PY37: {
166 Feature.UNICODE_LITERALS,
168 Feature.NUMERIC_UNDERSCORES,
169 Feature.TRAILING_COMMA_IN_CALL,
170 Feature.TRAILING_COMMA_IN_DEF,
171 Feature.ASYNC_KEYWORDS,
173 TargetVersion.PY38: {
174 Feature.UNICODE_LITERALS,
176 Feature.NUMERIC_UNDERSCORES,
177 Feature.TRAILING_COMMA_IN_CALL,
178 Feature.TRAILING_COMMA_IN_DEF,
179 Feature.ASYNC_KEYWORDS,
180 Feature.ASSIGNMENT_EXPRESSIONS,
181 Feature.POS_ONLY_ARGUMENTS,
188 target_versions: Set[TargetVersion] = Factory(set)
189 line_length: int = DEFAULT_LINE_LENGTH
190 string_normalization: bool = True
193 def get_cache_key(self) -> str:
194 if self.target_versions:
195 version_str = ",".join(
197 for version in sorted(self.target_versions, key=lambda v: v.value)
203 str(self.line_length),
204 str(int(self.string_normalization)),
205 str(int(self.is_pyi)),
207 return ".".join(parts)
210 def supports_feature(target_versions: Set[TargetVersion], feature: Feature) -> bool:
211 return all(feature in VERSION_TO_FEATURES[version] for version in target_versions)
214 def read_pyproject_toml(
215 ctx: click.Context, param: click.Parameter, value: Union[str, int, bool, None]
217 """Inject Black configuration from "pyproject.toml" into defaults in `ctx`.
219 Returns the path to a successfully found and read configuration file, None
222 assert not isinstance(value, (int, bool)), "Invalid parameter type passed"
224 root = find_project_root(ctx.params.get("src", ()))
225 path = root / "pyproject.toml"
232 pyproject_toml = toml.load(value)
233 config = pyproject_toml.get("tool", {}).get("black", {})
234 except (toml.TomlDecodeError, OSError) as e:
235 raise click.FileError(
236 filename=value, hint=f"Error reading configuration file: {e}"
242 if ctx.default_map is None:
244 ctx.default_map.update( # type: ignore # bad types in .pyi
245 {k.replace("--", "").replace("-", "_"): v for k, v in config.items()}
250 @click.command(context_settings=dict(help_option_names=["-h", "--help"]))
251 @click.option("-c", "--code", type=str, help="Format the code passed in as a string.")
256 default=DEFAULT_LINE_LENGTH,
257 help="How many characters per line to allow.",
263 type=click.Choice([v.name.lower() for v in TargetVersion]),
264 callback=lambda c, p, v: [TargetVersion[val.upper()] for val in v],
267 "Python versions that should be supported by Black's output. [default: "
268 "per-file auto-detection]"
275 "Allow using Python 3.6-only syntax on all input files. This will put "
276 "trailing commas in function signatures and calls also after *args and "
277 "**kwargs. Deprecated; use --target-version instead. "
278 "[default: per-file auto-detection]"
285 "Format all input files like typing stubs regardless of file extension "
286 "(useful when piping source on standard input)."
291 "--skip-string-normalization",
293 help="Don't normalize string quotes or prefixes.",
299 "Don't write the files back, just return the status. Return code 0 "
300 "means nothing would change. Return code 1 means some files would be "
301 "reformatted. Return code 123 means there was an internal error."
307 help="Don't write the files back, just output a diff for each file on stdout.",
312 help="If --fast given, skip temporary sanity checks. [default: --safe]",
317 default=DEFAULT_INCLUDES,
319 "A regular expression that matches files and directories that should be "
320 "included on recursive searches. An empty value means all files are "
321 "included regardless of the name. Use forward slashes for directories on "
322 "all platforms (Windows, too). Exclusions are calculated first, inclusions "
330 default=DEFAULT_EXCLUDES,
332 "A regular expression that matches files and directories that should be "
333 "excluded on recursive searches. An empty value means no paths are excluded. "
334 "Use forward slashes for directories on all platforms (Windows, too). "
335 "Exclusions are calculated first, inclusions later."
344 "Don't emit non-error messages to stderr. Errors are still emitted; "
345 "silence those with 2>/dev/null."
353 "Also emit messages to stderr about files that were not changed or were "
354 "ignored due to --exclude=."
357 @click.version_option(version=__version__)
362 exists=True, file_okay=True, dir_okay=True, readable=True, allow_dash=True
369 exists=False, file_okay=True, dir_okay=False, readable=True, allow_dash=False
372 callback=read_pyproject_toml,
373 help="Read configuration from PATH.",
380 target_version: List[TargetVersion],
386 skip_string_normalization: bool,
392 config: Optional[str],
394 """The uncompromising code formatter."""
395 write_back = WriteBack.from_configuration(check=check, diff=diff)
398 err(f"Cannot use both --target-version and --py36")
401 versions = set(target_version)
404 "--py36 is deprecated and will be removed in a future version. "
405 "Use --target-version py36 instead."
407 versions = PY36_VERSIONS
409 # We'll autodetect later.
412 target_versions=versions,
413 line_length=line_length,
415 string_normalization=not skip_string_normalization,
417 if config and verbose:
418 out(f"Using configuration from {config}.", bold=False, fg="blue")
420 print(format_str(code, mode=mode))
423 include_regex = re_compile_maybe_verbose(include)
425 err(f"Invalid regular expression for include given: {include!r}")
428 exclude_regex = re_compile_maybe_verbose(exclude)
430 err(f"Invalid regular expression for exclude given: {exclude!r}")
432 report = Report(check=check, quiet=quiet, verbose=verbose)
433 root = find_project_root(src)
434 sources: Set[Path] = set()
435 path_empty(src, quiet, verbose, ctx)
440 gen_python_files_in_dir(
441 p, root, include_regex, exclude_regex, report, get_gitignore(root)
444 elif p.is_file() or s == "-":
445 # if a file was explicitly given, we don't care about its extension
448 err(f"invalid path: {s}")
449 if len(sources) == 0:
450 if verbose or not quiet:
451 out("No Python files are present to be formatted. Nothing to do 😴")
454 if len(sources) == 1:
458 write_back=write_back,
464 sources=sources, fast=fast, write_back=write_back, mode=mode, report=report
467 if verbose or not quiet:
468 out("Oh no! 💥 💔 💥" if report.return_code else "All done! ✨ 🍰 ✨")
469 click.secho(str(report), err=True)
470 ctx.exit(report.return_code)
473 def path_empty(src: Tuple[str], quiet: bool, verbose: bool, ctx: click.Context) -> None:
475 Exit if there is no `src` provided for formatting
478 if verbose or not quiet:
479 out("No Path provided. Nothing to do 😴")
484 src: Path, fast: bool, write_back: WriteBack, mode: FileMode, report: "Report"
486 """Reformat a single file under `src` without spawning child processes.
488 `fast`, `write_back`, and `mode` options are passed to
489 :func:`format_file_in_place` or :func:`format_stdin_to_stdout`.
493 if not src.is_file() and str(src) == "-":
494 if format_stdin_to_stdout(fast=fast, write_back=write_back, mode=mode):
495 changed = Changed.YES
498 if write_back != WriteBack.DIFF:
499 cache = read_cache(mode)
500 res_src = src.resolve()
501 if res_src in cache and cache[res_src] == get_cache_info(res_src):
502 changed = Changed.CACHED
503 if changed is not Changed.CACHED and format_file_in_place(
504 src, fast=fast, write_back=write_back, mode=mode
506 changed = Changed.YES
507 if (write_back is WriteBack.YES and changed is not Changed.CACHED) or (
508 write_back is WriteBack.CHECK and changed is Changed.NO
510 write_cache(cache, [src], mode)
511 report.done(src, changed)
512 except Exception as exc:
513 report.failed(src, str(exc))
519 write_back: WriteBack,
523 """Reformat multiple files using a ProcessPoolExecutor."""
524 loop = asyncio.get_event_loop()
525 worker_count = os.cpu_count()
526 if sys.platform == "win32":
527 # Work around https://bugs.python.org/issue26903
528 worker_count = min(worker_count, 61)
529 executor = ProcessPoolExecutor(max_workers=worker_count)
531 loop.run_until_complete(
535 write_back=write_back,
547 async def schedule_formatting(
550 write_back: WriteBack,
553 loop: asyncio.AbstractEventLoop,
556 """Run formatting of `sources` in parallel using the provided `executor`.
558 (Use ProcessPoolExecutors for actual parallelism.)
560 `write_back`, `fast`, and `mode` options are passed to
561 :func:`format_file_in_place`.
564 if write_back != WriteBack.DIFF:
565 cache = read_cache(mode)
566 sources, cached = filter_cached(cache, sources)
567 for src in sorted(cached):
568 report.done(src, Changed.CACHED)
573 sources_to_cache = []
575 if write_back == WriteBack.DIFF:
576 # For diff output, we need locks to ensure we don't interleave output
577 # from different processes.
579 lock = manager.Lock()
581 asyncio.ensure_future(
582 loop.run_in_executor(
583 executor, format_file_in_place, src, fast, mode, write_back, lock
586 for src in sorted(sources)
588 pending: Iterable[asyncio.Future] = tasks.keys()
590 loop.add_signal_handler(signal.SIGINT, cancel, pending)
591 loop.add_signal_handler(signal.SIGTERM, cancel, pending)
592 except NotImplementedError:
593 # There are no good alternatives for these on Windows.
596 done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
598 src = tasks.pop(task)
600 cancelled.append(task)
601 elif task.exception():
602 report.failed(src, str(task.exception()))
604 changed = Changed.YES if task.result() else Changed.NO
605 # If the file was written back or was successfully checked as
606 # well-formatted, store this information in the cache.
607 if write_back is WriteBack.YES or (
608 write_back is WriteBack.CHECK and changed is Changed.NO
610 sources_to_cache.append(src)
611 report.done(src, changed)
613 await asyncio.gather(*cancelled, loop=loop, return_exceptions=True)
615 write_cache(cache, sources_to_cache, mode)
618 def format_file_in_place(
622 write_back: WriteBack = WriteBack.NO,
623 lock: Any = None, # multiprocessing.Manager().Lock() is some crazy proxy
625 """Format file under `src` path. Return True if changed.
627 If `write_back` is DIFF, write a diff to stdout. If it is YES, write reformatted
629 `mode` and `fast` options are passed to :func:`format_file_contents`.
631 if src.suffix == ".pyi":
632 mode = evolve(mode, is_pyi=True)
634 then = datetime.utcfromtimestamp(src.stat().st_mtime)
635 with open(src, "rb") as buf:
636 src_contents, encoding, newline = decode_bytes(buf.read())
638 dst_contents = format_file_contents(src_contents, fast=fast, mode=mode)
639 except NothingChanged:
642 if write_back == write_back.YES:
643 with open(src, "w", encoding=encoding, newline=newline) as f:
644 f.write(dst_contents)
645 elif write_back == write_back.DIFF:
646 now = datetime.utcnow()
647 src_name = f"{src}\t{then} +0000"
648 dst_name = f"{src}\t{now} +0000"
649 diff_contents = diff(src_contents, dst_contents, src_name, dst_name)
651 with lock or nullcontext():
652 f = io.TextIOWrapper(
658 f.write(diff_contents)
664 def format_stdin_to_stdout(
665 fast: bool, *, write_back: WriteBack = WriteBack.NO, mode: FileMode
667 """Format file on stdin. Return True if changed.
669 If `write_back` is YES, write reformatted code back to stdout. If it is DIFF,
670 write a diff to stdout. The `mode` argument is passed to
671 :func:`format_file_contents`.
673 then = datetime.utcnow()
674 src, encoding, newline = decode_bytes(sys.stdin.buffer.read())
677 dst = format_file_contents(src, fast=fast, mode=mode)
680 except NothingChanged:
684 f = io.TextIOWrapper(
685 sys.stdout.buffer, encoding=encoding, newline=newline, write_through=True
687 if write_back == WriteBack.YES:
689 elif write_back == WriteBack.DIFF:
690 now = datetime.utcnow()
691 src_name = f"STDIN\t{then} +0000"
692 dst_name = f"STDOUT\t{now} +0000"
693 f.write(diff(src, dst, src_name, dst_name))
697 def format_file_contents(
698 src_contents: str, *, fast: bool, mode: FileMode
700 """Reformat contents a file and return new contents.
702 If `fast` is False, additionally confirm that the reformatted code is
703 valid by calling :func:`assert_equivalent` and :func:`assert_stable` on it.
704 `mode` is passed to :func:`format_str`.
706 if src_contents.strip() == "":
709 dst_contents = format_str(src_contents, mode=mode)
710 if src_contents == dst_contents:
714 assert_equivalent(src_contents, dst_contents)
715 assert_stable(src_contents, dst_contents, mode=mode)
719 def format_str(src_contents: str, *, mode: FileMode) -> FileContent:
720 """Reformat a string and return new contents.
722 `mode` determines formatting options, such as how many characters per line are
725 src_node = lib2to3_parse(src_contents.lstrip(), mode.target_versions)
727 future_imports = get_future_imports(src_node)
728 if mode.target_versions:
729 versions = mode.target_versions
731 versions = detect_target_versions(src_node)
732 normalize_fmt_off(src_node)
733 lines = LineGenerator(
734 remove_u_prefix="unicode_literals" in future_imports
735 or supports_feature(versions, Feature.UNICODE_LITERALS),
737 normalize_strings=mode.string_normalization,
739 elt = EmptyLineTracker(is_pyi=mode.is_pyi)
742 split_line_features = {
744 for feature in {Feature.TRAILING_COMMA_IN_CALL, Feature.TRAILING_COMMA_IN_DEF}
745 if supports_feature(versions, feature)
747 for current_line in lines.visit(src_node):
748 for _ in range(after):
749 dst_contents.append(str(empty_line))
750 before, after = elt.maybe_empty_lines(current_line)
751 for _ in range(before):
752 dst_contents.append(str(empty_line))
753 for line in split_line(
754 current_line, line_length=mode.line_length, features=split_line_features
756 dst_contents.append(str(line))
757 return "".join(dst_contents)
760 def decode_bytes(src: bytes) -> Tuple[FileContent, Encoding, NewLine]:
761 """Return a tuple of (decoded_contents, encoding, newline).
763 `newline` is either CRLF or LF but `decoded_contents` is decoded with
764 universal newlines (i.e. only contains LF).
766 srcbuf = io.BytesIO(src)
767 encoding, lines = tokenize.detect_encoding(srcbuf.readline)
769 return "", encoding, "\n"
771 newline = "\r\n" if b"\r\n" == lines[0][-2:] else "\n"
773 with io.TextIOWrapper(srcbuf, encoding) as tiow:
774 return tiow.read(), encoding, newline
777 def get_grammars(target_versions: Set[TargetVersion]) -> List[Grammar]:
778 if not target_versions:
779 # No target_version specified, so try all grammars.
782 pygram.python_grammar_no_print_statement_no_exec_statement_async_keywords,
784 pygram.python_grammar_no_print_statement_no_exec_statement,
785 # Python 2.7 with future print_function import
786 pygram.python_grammar_no_print_statement,
788 pygram.python_grammar,
791 if all(version.is_python2() for version in target_versions):
792 # Python 2-only code, so try Python 2 grammars.
794 # Python 2.7 with future print_function import
795 pygram.python_grammar_no_print_statement,
797 pygram.python_grammar,
800 # Python 3-compatible code, so only try Python 3 grammar.
802 # If we have to parse both, try to parse async as a keyword first
803 if not supports_feature(target_versions, Feature.ASYNC_IDENTIFIERS):
806 pygram.python_grammar_no_print_statement_no_exec_statement_async_keywords
808 if not supports_feature(target_versions, Feature.ASYNC_KEYWORDS):
810 grammars.append(pygram.python_grammar_no_print_statement_no_exec_statement)
811 # At least one of the above branches must have been taken, because every Python
812 # version has exactly one of the two 'ASYNC_*' flags
816 def lib2to3_parse(src_txt: str, target_versions: Iterable[TargetVersion] = ()) -> Node:
817 """Given a string with source, return the lib2to3 Node."""
818 if src_txt[-1:] != "\n":
821 for grammar in get_grammars(set(target_versions)):
822 drv = driver.Driver(grammar, pytree.convert)
824 result = drv.parse_string(src_txt, True)
827 except ParseError as pe:
828 lineno, column = pe.context[1]
829 lines = src_txt.splitlines()
831 faulty_line = lines[lineno - 1]
833 faulty_line = "<line number missing in source>"
834 exc = InvalidInput(f"Cannot parse: {lineno}:{column}: {faulty_line}")
838 if isinstance(result, Leaf):
839 result = Node(syms.file_input, [result])
843 def lib2to3_unparse(node: Node) -> str:
844 """Given a lib2to3 node, return its string representation."""
852 class Visitor(Generic[T]):
853 """Basic lib2to3 visitor that yields things of type `T` on `visit()`."""
855 def visit(self, node: LN) -> Iterator[T]:
856 """Main method to visit `node` and its children.
858 It tries to find a `visit_*()` method for the given `node.type`, like
859 `visit_simple_stmt` for Node objects or `visit_INDENT` for Leaf objects.
860 If no dedicated `visit_*()` method is found, chooses `visit_default()`
863 Then yields objects of type `T` from the selected visitor.
866 name = token.tok_name[node.type]
868 name = type_repr(node.type)
869 yield from getattr(self, f"visit_{name}", self.visit_default)(node)
871 def visit_default(self, node: LN) -> Iterator[T]:
872 """Default `visit_*()` implementation. Recurses to children of `node`."""
873 if isinstance(node, Node):
874 for child in node.children:
875 yield from self.visit(child)
879 class DebugVisitor(Visitor[T]):
882 def visit_default(self, node: LN) -> Iterator[T]:
883 indent = " " * (2 * self.tree_depth)
884 if isinstance(node, Node):
885 _type = type_repr(node.type)
886 out(f"{indent}{_type}", fg="yellow")
888 for child in node.children:
889 yield from self.visit(child)
892 out(f"{indent}/{_type}", fg="yellow", bold=False)
894 _type = token.tok_name.get(node.type, str(node.type))
895 out(f"{indent}{_type}", fg="blue", nl=False)
897 # We don't have to handle prefixes for `Node` objects since
898 # that delegates to the first child anyway.
899 out(f" {node.prefix!r}", fg="green", bold=False, nl=False)
900 out(f" {node.value!r}", fg="blue", bold=False)
903 def show(cls, code: Union[str, Leaf, Node]) -> None:
904 """Pretty-print the lib2to3 AST of a given string of `code`.
906 Convenience method for debugging.
908 v: DebugVisitor[None] = DebugVisitor()
909 if isinstance(code, str):
910 code = lib2to3_parse(code)
914 WHITESPACE = {token.DEDENT, token.INDENT, token.NEWLINE}
925 STANDALONE_COMMENT = 153
926 token.tok_name[STANDALONE_COMMENT] = "STANDALONE_COMMENT"
927 LOGIC_OPERATORS = {"and", "or"}
952 STARS = {token.STAR, token.DOUBLESTAR}
953 VARARGS_SPECIALS = STARS | {token.SLASH}
956 syms.argument, # double star in arglist
957 syms.trailer, # single argument to call
959 syms.varargslist, # lambdas
961 UNPACKING_PARENTS = {
962 syms.atom, # single element of a list or set literal
966 syms.testlist_star_expr,
1001 COMPREHENSION_PRIORITY = 20
1003 TERNARY_PRIORITY = 16
1005 STRING_PRIORITY = 12
1006 COMPARATOR_PRIORITY = 10
1009 token.CIRCUMFLEX: 8,
1012 token.RIGHTSHIFT: 6,
1017 token.DOUBLESLASH: 4,
1021 token.DOUBLESTAR: 2,
1027 class BracketTracker:
1028 """Keeps track of brackets on a line."""
1031 bracket_match: Dict[Tuple[Depth, NodeType], Leaf] = Factory(dict)
1032 delimiters: Dict[LeafID, Priority] = Factory(dict)
1033 previous: Optional[Leaf] = None
1034 _for_loop_depths: List[int] = Factory(list)
1035 _lambda_argument_depths: List[int] = Factory(list)
1037 def mark(self, leaf: Leaf) -> None:
1038 """Mark `leaf` with bracket-related metadata. Keep track of delimiters.
1040 All leaves receive an int `bracket_depth` field that stores how deep
1041 within brackets a given leaf is. 0 means there are no enclosing brackets
1042 that started on this line.
1044 If a leaf is itself a closing bracket, it receives an `opening_bracket`
1045 field that it forms a pair with. This is a one-directional link to
1046 avoid reference cycles.
1048 If a leaf is a delimiter (a token on which Black can split the line if
1049 needed) and it's on depth 0, its `id()` is stored in the tracker's
1052 if leaf.type == token.COMMENT:
1055 self.maybe_decrement_after_for_loop_variable(leaf)
1056 self.maybe_decrement_after_lambda_arguments(leaf)
1057 if leaf.type in CLOSING_BRACKETS:
1059 opening_bracket = self.bracket_match.pop((self.depth, leaf.type))
1060 leaf.opening_bracket = opening_bracket
1061 leaf.bracket_depth = self.depth
1063 delim = is_split_before_delimiter(leaf, self.previous)
1064 if delim and self.previous is not None:
1065 self.delimiters[id(self.previous)] = delim
1067 delim = is_split_after_delimiter(leaf, self.previous)
1069 self.delimiters[id(leaf)] = delim
1070 if leaf.type in OPENING_BRACKETS:
1071 self.bracket_match[self.depth, BRACKET[leaf.type]] = leaf
1073 self.previous = leaf
1074 self.maybe_increment_lambda_arguments(leaf)
1075 self.maybe_increment_for_loop_variable(leaf)
1077 def any_open_brackets(self) -> bool:
1078 """Return True if there is an yet unmatched open bracket on the line."""
1079 return bool(self.bracket_match)
1081 def max_delimiter_priority(self, exclude: Iterable[LeafID] = ()) -> Priority:
1082 """Return the highest priority of a delimiter found on the line.
1084 Values are consistent with what `is_split_*_delimiter()` return.
1085 Raises ValueError on no delimiters.
1087 return max(v for k, v in self.delimiters.items() if k not in exclude)
1089 def delimiter_count_with_priority(self, priority: Priority = 0) -> int:
1090 """Return the number of delimiters with the given `priority`.
1092 If no `priority` is passed, defaults to max priority on the line.
1094 if not self.delimiters:
1097 priority = priority or self.max_delimiter_priority()
1098 return sum(1 for p in self.delimiters.values() if p == priority)
1100 def maybe_increment_for_loop_variable(self, leaf: Leaf) -> bool:
1101 """In a for loop, or comprehension, the variables are often unpacks.
1103 To avoid splitting on the comma in this situation, increase the depth of
1104 tokens between `for` and `in`.
1106 if leaf.type == token.NAME and leaf.value == "for":
1108 self._for_loop_depths.append(self.depth)
1113 def maybe_decrement_after_for_loop_variable(self, leaf: Leaf) -> bool:
1114 """See `maybe_increment_for_loop_variable` above for explanation."""
1116 self._for_loop_depths
1117 and self._for_loop_depths[-1] == self.depth
1118 and leaf.type == token.NAME
1119 and leaf.value == "in"
1122 self._for_loop_depths.pop()
1127 def maybe_increment_lambda_arguments(self, leaf: Leaf) -> bool:
1128 """In a lambda expression, there might be more than one argument.
1130 To avoid splitting on the comma in this situation, increase the depth of
1131 tokens between `lambda` and `:`.
1133 if leaf.type == token.NAME and leaf.value == "lambda":
1135 self._lambda_argument_depths.append(self.depth)
1140 def maybe_decrement_after_lambda_arguments(self, leaf: Leaf) -> bool:
1141 """See `maybe_increment_lambda_arguments` above for explanation."""
1143 self._lambda_argument_depths
1144 and self._lambda_argument_depths[-1] == self.depth
1145 and leaf.type == token.COLON
1148 self._lambda_argument_depths.pop()
1153 def get_open_lsqb(self) -> Optional[Leaf]:
1154 """Return the most recent opening square bracket (if any)."""
1155 return self.bracket_match.get((self.depth - 1, token.RSQB))
1160 """Holds leaves and comments. Can be printed with `str(line)`."""
1163 leaves: List[Leaf] = Factory(list)
1164 comments: Dict[LeafID, List[Leaf]] = Factory(dict) # keys ordered like `leaves`
1165 bracket_tracker: BracketTracker = Factory(BracketTracker)
1166 inside_brackets: bool = False
1167 should_explode: bool = False
1169 def append(self, leaf: Leaf, preformatted: bool = False) -> None:
1170 """Add a new `leaf` to the end of the line.
1172 Unless `preformatted` is True, the `leaf` will receive a new consistent
1173 whitespace prefix and metadata applied by :class:`BracketTracker`.
1174 Trailing commas are maybe removed, unpacked for loop variables are
1175 demoted from being delimiters.
1177 Inline comments are put aside.
1179 has_value = leaf.type in BRACKETS or bool(leaf.value.strip())
1183 if token.COLON == leaf.type and self.is_class_paren_empty:
1184 del self.leaves[-2:]
1185 if self.leaves and not preformatted:
1186 # Note: at this point leaf.prefix should be empty except for
1187 # imports, for which we only preserve newlines.
1188 leaf.prefix += whitespace(
1189 leaf, complex_subscript=self.is_complex_subscript(leaf)
1191 if self.inside_brackets or not preformatted:
1192 self.bracket_tracker.mark(leaf)
1193 self.maybe_remove_trailing_comma(leaf)
1194 if not self.append_comment(leaf):
1195 self.leaves.append(leaf)
1197 def append_safe(self, leaf: Leaf, preformatted: bool = False) -> None:
1198 """Like :func:`append()` but disallow invalid standalone comment structure.
1200 Raises ValueError when any `leaf` is appended after a standalone comment
1201 or when a standalone comment is not the first leaf on the line.
1203 if self.bracket_tracker.depth == 0:
1205 raise ValueError("cannot append to standalone comments")
1207 if self.leaves and leaf.type == STANDALONE_COMMENT:
1209 "cannot append standalone comments to a populated line"
1212 self.append(leaf, preformatted=preformatted)
1215 def is_comment(self) -> bool:
1216 """Is this line a standalone comment?"""
1217 return len(self.leaves) == 1 and self.leaves[0].type == STANDALONE_COMMENT
1220 def is_decorator(self) -> bool:
1221 """Is this line a decorator?"""
1222 return bool(self) and self.leaves[0].type == token.AT
1225 def is_import(self) -> bool:
1226 """Is this an import line?"""
1227 return bool(self) and is_import(self.leaves[0])
1230 def is_class(self) -> bool:
1231 """Is this line a class definition?"""
1234 and self.leaves[0].type == token.NAME
1235 and self.leaves[0].value == "class"
1239 def is_stub_class(self) -> bool:
1240 """Is this line a class definition with a body consisting only of "..."?"""
1241 return self.is_class and self.leaves[-3:] == [
1242 Leaf(token.DOT, ".") for _ in range(3)
1246 def is_collection_with_optional_trailing_comma(self) -> bool:
1247 """Is this line a collection literal with a trailing comma that's optional?
1249 Note that the trailing comma in a 1-tuple is not optional.
1251 if not self.leaves or len(self.leaves) < 4:
1254 # Look for and address a trailing colon.
1255 if self.leaves[-1].type == token.COLON:
1256 closer = self.leaves[-2]
1259 closer = self.leaves[-1]
1261 if closer.type not in CLOSING_BRACKETS or self.inside_brackets:
1264 if closer.type == token.RPAR:
1265 # Tuples require an extra check, because if there's only
1266 # one element in the tuple removing the comma unmakes the
1269 # We also check for parens before looking for the trailing
1270 # comma because in some cases (eg assigning a dict
1271 # literal) the literal gets wrapped in temporary parens
1272 # during parsing. This case is covered by the
1273 # collections.py test data.
1274 opener = closer.opening_bracket
1275 for _open_index, leaf in enumerate(self.leaves):
1280 # Couldn't find the matching opening paren, play it safe.
1284 comma_depth = self.leaves[close_index - 1].bracket_depth
1285 for leaf in self.leaves[_open_index + 1 : close_index]:
1286 if leaf.bracket_depth == comma_depth and leaf.type == token.COMMA:
1289 # We haven't looked yet for the trailing comma because
1290 # we might also have caught noop parens.
1291 return self.leaves[close_index - 1].type == token.COMMA
1294 return False # it's either a one-tuple or didn't have a trailing comma
1296 if self.leaves[close_index - 1].type in CLOSING_BRACKETS:
1298 closer = self.leaves[close_index]
1299 if closer.type == token.RPAR:
1300 # TODO: this is a gut feeling. Will we ever see this?
1303 if self.leaves[close_index - 1].type != token.COMMA:
1309 def is_def(self) -> bool:
1310 """Is this a function definition? (Also returns True for async defs.)"""
1312 first_leaf = self.leaves[0]
1317 second_leaf: Optional[Leaf] = self.leaves[1]
1320 return (first_leaf.type == token.NAME and first_leaf.value == "def") or (
1321 first_leaf.type == token.ASYNC
1322 and second_leaf is not None
1323 and second_leaf.type == token.NAME
1324 and second_leaf.value == "def"
1328 def is_class_paren_empty(self) -> bool:
1329 """Is this a class with no base classes but using parentheses?
1331 Those are unnecessary and should be removed.
1335 and len(self.leaves) == 4
1337 and self.leaves[2].type == token.LPAR
1338 and self.leaves[2].value == "("
1339 and self.leaves[3].type == token.RPAR
1340 and self.leaves[3].value == ")"
1344 def is_triple_quoted_string(self) -> bool:
1345 """Is the line a triple quoted string?"""
1348 and self.leaves[0].type == token.STRING
1349 and self.leaves[0].value.startswith(('"""', "'''"))
1352 def contains_standalone_comments(self, depth_limit: int = sys.maxsize) -> bool:
1353 """If so, needs to be split before emitting."""
1354 for leaf in self.leaves:
1355 if leaf.type == STANDALONE_COMMENT and leaf.bracket_depth <= depth_limit:
1360 def contains_uncollapsable_type_comments(self) -> bool:
1363 last_leaf = self.leaves[-1]
1364 ignored_ids.add(id(last_leaf))
1365 if last_leaf.type == token.COMMA or (
1366 last_leaf.type == token.RPAR and not last_leaf.value
1368 # When trailing commas or optional parens are inserted by Black for
1369 # consistency, comments after the previous last element are not moved
1370 # (they don't have to, rendering will still be correct). So we ignore
1371 # trailing commas and invisible.
1372 last_leaf = self.leaves[-2]
1373 ignored_ids.add(id(last_leaf))
1377 # A type comment is uncollapsable if it is attached to a leaf
1378 # that isn't at the end of the line (since that could cause it
1379 # to get associated to a different argument) or if there are
1380 # comments before it (since that could cause it to get hidden
1382 comment_seen = False
1383 for leaf_id, comments in self.comments.items():
1384 for comment in comments:
1385 if is_type_comment(comment):
1386 if leaf_id not in ignored_ids or comment_seen:
1393 def contains_unsplittable_type_ignore(self) -> bool:
1397 # If a 'type: ignore' is attached to the end of a line, we
1398 # can't split the line, because we can't know which of the
1399 # subexpressions the ignore was meant to apply to.
1401 # We only want this to apply to actual physical lines from the
1402 # original source, though: we don't want the presence of a
1403 # 'type: ignore' at the end of a multiline expression to
1404 # justify pushing it all onto one line. Thus we
1405 # (unfortunately) need to check the actual source lines and
1406 # only report an unsplittable 'type: ignore' if this line was
1407 # one line in the original code.
1409 # Grab the first and last line numbers, skipping generated leaves
1410 first_line = next((l.lineno for l in self.leaves if l.lineno != 0), 0)
1411 last_line = next((l.lineno for l in reversed(self.leaves) if l.lineno != 0), 0)
1413 if first_line == last_line:
1414 # We look at the last two leaves since a comma or an
1415 # invisible paren could have been added at the end of the
1417 for node in self.leaves[-2:]:
1418 for comment in self.comments.get(id(node), []):
1419 if is_type_comment(comment, " ignore"):
1424 def contains_multiline_strings(self) -> bool:
1425 for leaf in self.leaves:
1426 if is_multiline_string(leaf):
1431 def maybe_remove_trailing_comma(self, closing: Leaf) -> bool:
1432 """Remove trailing comma if there is one and it's safe."""
1433 if not (self.leaves and self.leaves[-1].type == token.COMMA):
1436 # We remove trailing commas only in the case of importing a
1437 # single name from a module.
1441 and len(self.leaves) > 4
1442 and self.leaves[-1].type == token.COMMA
1443 and closing.type in CLOSING_BRACKETS
1444 and self.leaves[-4].type == token.NAME
1446 # regular `from foo import bar,`
1447 self.leaves[-4].value == "import"
1448 # `from foo import (bar as baz,)
1450 len(self.leaves) > 6
1451 and self.leaves[-6].value == "import"
1452 and self.leaves[-3].value == "as"
1454 # `from foo import bar as baz,`
1456 len(self.leaves) > 5
1457 and self.leaves[-5].value == "import"
1458 and self.leaves[-3].value == "as"
1461 and closing.type == token.RPAR
1465 self.remove_trailing_comma()
1468 def append_comment(self, comment: Leaf) -> bool:
1469 """Add an inline or standalone comment to the line."""
1471 comment.type == STANDALONE_COMMENT
1472 and self.bracket_tracker.any_open_brackets()
1477 if comment.type != token.COMMENT:
1481 comment.type = STANDALONE_COMMENT
1485 last_leaf = self.leaves[-1]
1487 last_leaf.type == token.RPAR
1488 and not last_leaf.value
1489 and last_leaf.parent
1490 and len(list(last_leaf.parent.leaves())) <= 3
1491 and not is_type_comment(comment)
1493 # Comments on an optional parens wrapping a single leaf should belong to
1494 # the wrapped node except if it's a type comment. Pinning the comment like
1495 # this avoids unstable formatting caused by comment migration.
1496 if len(self.leaves) < 2:
1497 comment.type = STANDALONE_COMMENT
1501 last_leaf = self.leaves[-2]
1502 self.comments.setdefault(id(last_leaf), []).append(comment)
1505 def comments_after(self, leaf: Leaf) -> List[Leaf]:
1506 """Generate comments that should appear directly after `leaf`."""
1507 return self.comments.get(id(leaf), [])
1509 def remove_trailing_comma(self) -> None:
1510 """Remove the trailing comma and moves the comments attached to it."""
1511 trailing_comma = self.leaves.pop()
1512 trailing_comma_comments = self.comments.pop(id(trailing_comma), [])
1513 self.comments.setdefault(id(self.leaves[-1]), []).extend(
1514 trailing_comma_comments
1517 def is_complex_subscript(self, leaf: Leaf) -> bool:
1518 """Return True iff `leaf` is part of a slice with non-trivial exprs."""
1519 open_lsqb = self.bracket_tracker.get_open_lsqb()
1520 if open_lsqb is None:
1523 subscript_start = open_lsqb.next_sibling
1525 if isinstance(subscript_start, Node):
1526 if subscript_start.type == syms.listmaker:
1529 if subscript_start.type == syms.subscriptlist:
1530 subscript_start = child_towards(subscript_start, leaf)
1531 return subscript_start is not None and any(
1532 n.type in TEST_DESCENDANTS for n in subscript_start.pre_order()
1535 def __str__(self) -> str:
1536 """Render the line."""
1540 indent = " " * self.depth
1541 leaves = iter(self.leaves)
1542 first = next(leaves)
1543 res = f"{first.prefix}{indent}{first.value}"
1546 for comment in itertools.chain.from_iterable(self.comments.values()):
1550 def __bool__(self) -> bool:
1551 """Return True if the line has leaves or comments."""
1552 return bool(self.leaves or self.comments)
1556 class EmptyLineTracker:
1557 """Provides a stateful method that returns the number of potential extra
1558 empty lines needed before and after the currently processed line.
1560 Note: this tracker works on lines that haven't been split yet. It assumes
1561 the prefix of the first leaf consists of optional newlines. Those newlines
1562 are consumed by `maybe_empty_lines()` and included in the computation.
1565 is_pyi: bool = False
1566 previous_line: Optional[Line] = None
1567 previous_after: int = 0
1568 previous_defs: List[int] = Factory(list)
1570 def maybe_empty_lines(self, current_line: Line) -> Tuple[int, int]:
1571 """Return the number of extra empty lines before and after the `current_line`.
1573 This is for separating `def`, `async def` and `class` with extra empty
1574 lines (two on module-level).
1576 before, after = self._maybe_empty_lines(current_line)
1578 # Black should not insert empty lines at the beginning
1581 if self.previous_line is None
1582 else before - self.previous_after
1584 self.previous_after = after
1585 self.previous_line = current_line
1586 return before, after
1588 def _maybe_empty_lines(self, current_line: Line) -> Tuple[int, int]:
1590 if current_line.depth == 0:
1591 max_allowed = 1 if self.is_pyi else 2
1592 if current_line.leaves:
1593 # Consume the first leaf's extra newlines.
1594 first_leaf = current_line.leaves[0]
1595 before = first_leaf.prefix.count("\n")
1596 before = min(before, max_allowed)
1597 first_leaf.prefix = ""
1600 depth = current_line.depth
1601 while self.previous_defs and self.previous_defs[-1] >= depth:
1602 self.previous_defs.pop()
1604 before = 0 if depth else 1
1606 before = 1 if depth else 2
1607 if current_line.is_decorator or current_line.is_def or current_line.is_class:
1608 return self._maybe_empty_lines_for_class_or_def(current_line, before)
1612 and self.previous_line.is_import
1613 and not current_line.is_import
1614 and depth == self.previous_line.depth
1616 return (before or 1), 0
1620 and self.previous_line.is_class
1621 and current_line.is_triple_quoted_string
1627 def _maybe_empty_lines_for_class_or_def(
1628 self, current_line: Line, before: int
1629 ) -> Tuple[int, int]:
1630 if not current_line.is_decorator:
1631 self.previous_defs.append(current_line.depth)
1632 if self.previous_line is None:
1633 # Don't insert empty lines before the first line in the file.
1636 if self.previous_line.is_decorator:
1639 if self.previous_line.depth < current_line.depth and (
1640 self.previous_line.is_class or self.previous_line.is_def
1645 self.previous_line.is_comment
1646 and self.previous_line.depth == current_line.depth
1652 if self.previous_line.depth > current_line.depth:
1654 elif current_line.is_class or self.previous_line.is_class:
1655 if current_line.is_stub_class and self.previous_line.is_stub_class:
1656 # No blank line between classes with an empty body
1660 elif current_line.is_def and not self.previous_line.is_def:
1661 # Blank line between a block of functions and a block of non-functions
1667 if current_line.depth and newlines:
1673 class LineGenerator(Visitor[Line]):
1674 """Generates reformatted Line objects. Empty lines are not emitted.
1676 Note: destroys the tree it's visiting by mutating prefixes of its leaves
1677 in ways that will no longer stringify to valid Python code on the tree.
1680 is_pyi: bool = False
1681 normalize_strings: bool = True
1682 current_line: Line = Factory(Line)
1683 remove_u_prefix: bool = False
1685 def line(self, indent: int = 0) -> Iterator[Line]:
1688 If the line is empty, only emit if it makes sense.
1689 If the line is too long, split it first and then generate.
1691 If any lines were generated, set up a new current_line.
1693 if not self.current_line:
1694 self.current_line.depth += indent
1695 return # Line is empty, don't emit. Creating a new one unnecessary.
1697 complete_line = self.current_line
1698 self.current_line = Line(depth=complete_line.depth + indent)
1701 def visit_default(self, node: LN) -> Iterator[Line]:
1702 """Default `visit_*()` implementation. Recurses to children of `node`."""
1703 if isinstance(node, Leaf):
1704 any_open_brackets = self.current_line.bracket_tracker.any_open_brackets()
1705 for comment in generate_comments(node):
1706 if any_open_brackets:
1707 # any comment within brackets is subject to splitting
1708 self.current_line.append(comment)
1709 elif comment.type == token.COMMENT:
1710 # regular trailing comment
1711 self.current_line.append(comment)
1712 yield from self.line()
1715 # regular standalone comment
1716 yield from self.line()
1718 self.current_line.append(comment)
1719 yield from self.line()
1721 normalize_prefix(node, inside_brackets=any_open_brackets)
1722 if self.normalize_strings and node.type == token.STRING:
1723 normalize_string_prefix(node, remove_u_prefix=self.remove_u_prefix)
1724 normalize_string_quotes(node)
1725 if node.type == token.NUMBER:
1726 normalize_numeric_literal(node)
1727 if node.type not in WHITESPACE:
1728 self.current_line.append(node)
1729 yield from super().visit_default(node)
1731 def visit_INDENT(self, node: Node) -> Iterator[Line]:
1732 """Increase indentation level, maybe yield a line."""
1733 # In blib2to3 INDENT never holds comments.
1734 yield from self.line(+1)
1735 yield from self.visit_default(node)
1737 def visit_DEDENT(self, node: Node) -> Iterator[Line]:
1738 """Decrease indentation level, maybe yield a line."""
1739 # The current line might still wait for trailing comments. At DEDENT time
1740 # there won't be any (they would be prefixes on the preceding NEWLINE).
1741 # Emit the line then.
1742 yield from self.line()
1744 # While DEDENT has no value, its prefix may contain standalone comments
1745 # that belong to the current indentation level. Get 'em.
1746 yield from self.visit_default(node)
1748 # Finally, emit the dedent.
1749 yield from self.line(-1)
1752 self, node: Node, keywords: Set[str], parens: Set[str]
1753 ) -> Iterator[Line]:
1754 """Visit a statement.
1756 This implementation is shared for `if`, `while`, `for`, `try`, `except`,
1757 `def`, `with`, `class`, `assert` and assignments.
1759 The relevant Python language `keywords` for a given statement will be
1760 NAME leaves within it. This methods puts those on a separate line.
1762 `parens` holds a set of string leaf values immediately after which
1763 invisible parens should be put.
1765 normalize_invisible_parens(node, parens_after=parens)
1766 for child in node.children:
1767 if child.type == token.NAME and child.value in keywords: # type: ignore
1768 yield from self.line()
1770 yield from self.visit(child)
1772 def visit_suite(self, node: Node) -> Iterator[Line]:
1773 """Visit a suite."""
1774 if self.is_pyi and is_stub_suite(node):
1775 yield from self.visit(node.children[2])
1777 yield from self.visit_default(node)
1779 def visit_simple_stmt(self, node: Node) -> Iterator[Line]:
1780 """Visit a statement without nested statements."""
1781 is_suite_like = node.parent and node.parent.type in STATEMENT
1783 if self.is_pyi and is_stub_body(node):
1784 yield from self.visit_default(node)
1786 yield from self.line(+1)
1787 yield from self.visit_default(node)
1788 yield from self.line(-1)
1791 if not self.is_pyi or not node.parent or not is_stub_suite(node.parent):
1792 yield from self.line()
1793 yield from self.visit_default(node)
1795 def visit_async_stmt(self, node: Node) -> Iterator[Line]:
1796 """Visit `async def`, `async for`, `async with`."""
1797 yield from self.line()
1799 children = iter(node.children)
1800 for child in children:
1801 yield from self.visit(child)
1803 if child.type == token.ASYNC:
1806 internal_stmt = next(children)
1807 for child in internal_stmt.children:
1808 yield from self.visit(child)
1810 def visit_decorators(self, node: Node) -> Iterator[Line]:
1811 """Visit decorators."""
1812 for child in node.children:
1813 yield from self.line()
1814 yield from self.visit(child)
1816 def visit_SEMI(self, leaf: Leaf) -> Iterator[Line]:
1817 """Remove a semicolon and put the other statement on a separate line."""
1818 yield from self.line()
1820 def visit_ENDMARKER(self, leaf: Leaf) -> Iterator[Line]:
1821 """End of file. Process outstanding comments and end with a newline."""
1822 yield from self.visit_default(leaf)
1823 yield from self.line()
1825 def visit_STANDALONE_COMMENT(self, leaf: Leaf) -> Iterator[Line]:
1826 if not self.current_line.bracket_tracker.any_open_brackets():
1827 yield from self.line()
1828 yield from self.visit_default(leaf)
1830 def visit_factor(self, node: Node) -> Iterator[Line]:
1831 """Force parentheses between a unary op and a binary power:
1833 -2 ** 8 -> -(2 ** 8)
1835 _operator, operand = node.children
1837 operand.type == syms.power
1838 and len(operand.children) == 3
1839 and operand.children[1].type == token.DOUBLESTAR
1841 lpar = Leaf(token.LPAR, "(")
1842 rpar = Leaf(token.RPAR, ")")
1843 index = operand.remove() or 0
1844 node.insert_child(index, Node(syms.atom, [lpar, operand, rpar]))
1845 yield from self.visit_default(node)
1847 def __attrs_post_init__(self) -> None:
1848 """You are in a twisty little maze of passages."""
1851 self.visit_assert_stmt = partial(v, keywords={"assert"}, parens={"assert", ","})
1852 self.visit_if_stmt = partial(
1853 v, keywords={"if", "else", "elif"}, parens={"if", "elif"}
1855 self.visit_while_stmt = partial(v, keywords={"while", "else"}, parens={"while"})
1856 self.visit_for_stmt = partial(v, keywords={"for", "else"}, parens={"for", "in"})
1857 self.visit_try_stmt = partial(
1858 v, keywords={"try", "except", "else", "finally"}, parens=Ø
1860 self.visit_except_clause = partial(v, keywords={"except"}, parens=Ø)
1861 self.visit_with_stmt = partial(v, keywords={"with"}, parens=Ø)
1862 self.visit_funcdef = partial(v, keywords={"def"}, parens=Ø)
1863 self.visit_classdef = partial(v, keywords={"class"}, parens=Ø)
1864 self.visit_expr_stmt = partial(v, keywords=Ø, parens=ASSIGNMENTS)
1865 self.visit_return_stmt = partial(v, keywords={"return"}, parens={"return"})
1866 self.visit_import_from = partial(v, keywords=Ø, parens={"import"})
1867 self.visit_del_stmt = partial(v, keywords=Ø, parens={"del"})
1868 self.visit_async_funcdef = self.visit_async_stmt
1869 self.visit_decorated = self.visit_decorators
1872 IMPLICIT_TUPLE = {syms.testlist, syms.testlist_star_expr, syms.exprlist}
1873 BRACKET = {token.LPAR: token.RPAR, token.LSQB: token.RSQB, token.LBRACE: token.RBRACE}
1874 OPENING_BRACKETS = set(BRACKET.keys())
1875 CLOSING_BRACKETS = set(BRACKET.values())
1876 BRACKETS = OPENING_BRACKETS | CLOSING_BRACKETS
1877 ALWAYS_NO_SPACE = CLOSING_BRACKETS | {token.COMMA, STANDALONE_COMMENT}
1880 def whitespace(leaf: Leaf, *, complex_subscript: bool) -> str: # noqa: C901
1881 """Return whitespace prefix if needed for the given `leaf`.
1883 `complex_subscript` signals whether the given leaf is part of a subscription
1884 which has non-trivial arguments, like arithmetic expressions or function calls.
1892 if t in ALWAYS_NO_SPACE:
1895 if t == token.COMMENT:
1898 assert p is not None, f"INTERNAL ERROR: hand-made leaf without parent: {leaf!r}"
1899 if t == token.COLON and p.type not in {
1906 prev = leaf.prev_sibling
1908 prevp = preceding_leaf(p)
1909 if not prevp or prevp.type in OPENING_BRACKETS:
1912 if t == token.COLON:
1913 if prevp.type == token.COLON:
1916 elif prevp.type != token.COMMA and not complex_subscript:
1921 if prevp.type == token.EQUAL:
1923 if prevp.parent.type in {
1931 elif prevp.parent.type == syms.typedargslist:
1932 # A bit hacky: if the equal sign has whitespace, it means we
1933 # previously found it's a typed argument. So, we're using
1937 elif prevp.type in VARARGS_SPECIALS:
1938 if is_vararg(prevp, within=VARARGS_PARENTS | UNPACKING_PARENTS):
1941 elif prevp.type == token.COLON:
1942 if prevp.parent and prevp.parent.type in {syms.subscript, syms.sliceop}:
1943 return SPACE if complex_subscript else NO
1947 and prevp.parent.type == syms.factor
1948 and prevp.type in MATH_OPERATORS
1953 prevp.type == token.RIGHTSHIFT
1955 and prevp.parent.type == syms.shift_expr
1956 and prevp.prev_sibling
1957 and prevp.prev_sibling.type == token.NAME
1958 and prevp.prev_sibling.value == "print" # type: ignore
1960 # Python 2 print chevron
1963 elif prev.type in OPENING_BRACKETS:
1966 if p.type in {syms.parameters, syms.arglist}:
1967 # untyped function signatures or calls
1968 if not prev or prev.type != token.COMMA:
1971 elif p.type == syms.varargslist:
1973 if prev and prev.type != token.COMMA:
1976 elif p.type == syms.typedargslist:
1977 # typed function signatures
1981 if t == token.EQUAL:
1982 if prev.type != syms.tname:
1985 elif prev.type == token.EQUAL:
1986 # A bit hacky: if the equal sign has whitespace, it means we
1987 # previously found it's a typed argument. So, we're using that, too.
1990 elif prev.type != token.COMMA:
1993 elif p.type == syms.tname:
1996 prevp = preceding_leaf(p)
1997 if not prevp or prevp.type != token.COMMA:
2000 elif p.type == syms.trailer:
2001 # attributes and calls
2002 if t == token.LPAR or t == token.RPAR:
2007 prevp = preceding_leaf(p)
2008 if not prevp or prevp.type != token.NUMBER:
2011 elif t == token.LSQB:
2014 elif prev.type != token.COMMA:
2017 elif p.type == syms.argument:
2019 if t == token.EQUAL:
2023 prevp = preceding_leaf(p)
2024 if not prevp or prevp.type == token.LPAR:
2027 elif prev.type in {token.EQUAL} | VARARGS_SPECIALS:
2030 elif p.type == syms.decorator:
2034 elif p.type == syms.dotted_name:
2038 prevp = preceding_leaf(p)
2039 if not prevp or prevp.type == token.AT or prevp.type == token.DOT:
2042 elif p.type == syms.classdef:
2046 if prev and prev.type == token.LPAR:
2049 elif p.type in {syms.subscript, syms.sliceop}:
2052 assert p.parent is not None, "subscripts are always parented"
2053 if p.parent.type == syms.subscriptlist:
2058 elif not complex_subscript:
2061 elif p.type == syms.atom:
2062 if prev and t == token.DOT:
2063 # dots, but not the first one.
2066 elif p.type == syms.dictsetmaker:
2068 if prev and prev.type == token.DOUBLESTAR:
2071 elif p.type in {syms.factor, syms.star_expr}:
2074 prevp = preceding_leaf(p)
2075 if not prevp or prevp.type in OPENING_BRACKETS:
2078 prevp_parent = prevp.parent
2079 assert prevp_parent is not None
2080 if prevp.type == token.COLON and prevp_parent.type in {
2086 elif prevp.type == token.EQUAL and prevp_parent.type == syms.argument:
2089 elif t in {token.NAME, token.NUMBER, token.STRING}:
2092 elif p.type == syms.import_from:
2094 if prev and prev.type == token.DOT:
2097 elif t == token.NAME:
2101 if prev and prev.type == token.DOT:
2104 elif p.type == syms.sliceop:
2110 def preceding_leaf(node: Optional[LN]) -> Optional[Leaf]:
2111 """Return the first leaf that precedes `node`, if any."""
2113 res = node.prev_sibling
2115 if isinstance(res, Leaf):
2119 return list(res.leaves())[-1]
2128 def child_towards(ancestor: Node, descendant: LN) -> Optional[LN]:
2129 """Return the child of `ancestor` that contains `descendant`."""
2130 node: Optional[LN] = descendant
2131 while node and node.parent != ancestor:
2136 def container_of(leaf: Leaf) -> LN:
2137 """Return `leaf` or one of its ancestors that is the topmost container of it.
2139 By "container" we mean a node where `leaf` is the very first child.
2141 same_prefix = leaf.prefix
2142 container: LN = leaf
2144 parent = container.parent
2148 if parent.children[0].prefix != same_prefix:
2151 if parent.type == syms.file_input:
2154 if parent.prev_sibling is not None and parent.prev_sibling.type in BRACKETS:
2161 def is_split_after_delimiter(leaf: Leaf, previous: Optional[Leaf] = None) -> Priority:
2162 """Return the priority of the `leaf` delimiter, given a line break after it.
2164 The delimiter priorities returned here are from those delimiters that would
2165 cause a line break after themselves.
2167 Higher numbers are higher priority.
2169 if leaf.type == token.COMMA:
2170 return COMMA_PRIORITY
2175 def is_split_before_delimiter(leaf: Leaf, previous: Optional[Leaf] = None) -> Priority:
2176 """Return the priority of the `leaf` delimiter, given a line break before it.
2178 The delimiter priorities returned here are from those delimiters that would
2179 cause a line break before themselves.
2181 Higher numbers are higher priority.
2183 if is_vararg(leaf, within=VARARGS_PARENTS | UNPACKING_PARENTS):
2184 # * and ** might also be MATH_OPERATORS but in this case they are not.
2185 # Don't treat them as a delimiter.
2189 leaf.type == token.DOT
2191 and leaf.parent.type not in {syms.import_from, syms.dotted_name}
2192 and (previous is None or previous.type in CLOSING_BRACKETS)
2197 leaf.type in MATH_OPERATORS
2199 and leaf.parent.type not in {syms.factor, syms.star_expr}
2201 return MATH_PRIORITIES[leaf.type]
2203 if leaf.type in COMPARATORS:
2204 return COMPARATOR_PRIORITY
2207 leaf.type == token.STRING
2208 and previous is not None
2209 and previous.type == token.STRING
2211 return STRING_PRIORITY
2213 if leaf.type not in {token.NAME, token.ASYNC}:
2219 and leaf.parent.type in {syms.comp_for, syms.old_comp_for}
2220 or leaf.type == token.ASYNC
2223 not isinstance(leaf.prev_sibling, Leaf)
2224 or leaf.prev_sibling.value != "async"
2226 return COMPREHENSION_PRIORITY
2231 and leaf.parent.type in {syms.comp_if, syms.old_comp_if}
2233 return COMPREHENSION_PRIORITY
2235 if leaf.value in {"if", "else"} and leaf.parent and leaf.parent.type == syms.test:
2236 return TERNARY_PRIORITY
2238 if leaf.value == "is":
2239 return COMPARATOR_PRIORITY
2244 and leaf.parent.type in {syms.comp_op, syms.comparison}
2246 previous is not None
2247 and previous.type == token.NAME
2248 and previous.value == "not"
2251 return COMPARATOR_PRIORITY
2256 and leaf.parent.type == syms.comp_op
2258 previous is not None
2259 and previous.type == token.NAME
2260 and previous.value == "is"
2263 return COMPARATOR_PRIORITY
2265 if leaf.value in LOGIC_OPERATORS and leaf.parent:
2266 return LOGIC_PRIORITY
2271 FMT_OFF = {"# fmt: off", "# fmt:off", "# yapf: disable"}
2272 FMT_ON = {"# fmt: on", "# fmt:on", "# yapf: enable"}
2275 def generate_comments(leaf: LN) -> Iterator[Leaf]:
2276 """Clean the prefix of the `leaf` and generate comments from it, if any.
2278 Comments in lib2to3 are shoved into the whitespace prefix. This happens
2279 in `pgen2/driver.py:Driver.parse_tokens()`. This was a brilliant implementation
2280 move because it does away with modifying the grammar to include all the
2281 possible places in which comments can be placed.
2283 The sad consequence for us though is that comments don't "belong" anywhere.
2284 This is why this function generates simple parentless Leaf objects for
2285 comments. We simply don't know what the correct parent should be.
2287 No matter though, we can live without this. We really only need to
2288 differentiate between inline and standalone comments. The latter don't
2289 share the line with any code.
2291 Inline comments are emitted as regular token.COMMENT leaves. Standalone
2292 are emitted with a fake STANDALONE_COMMENT token identifier.
2294 for pc in list_comments(leaf.prefix, is_endmarker=leaf.type == token.ENDMARKER):
2295 yield Leaf(pc.type, pc.value, prefix="\n" * pc.newlines)
2300 """Describes a piece of syntax that is a comment.
2302 It's not a :class:`blib2to3.pytree.Leaf` so that:
2304 * it can be cached (`Leaf` objects should not be reused more than once as
2305 they store their lineno, column, prefix, and parent information);
2306 * `newlines` and `consumed` fields are kept separate from the `value`. This
2307 simplifies handling of special marker comments like ``# fmt: off/on``.
2310 type: int # token.COMMENT or STANDALONE_COMMENT
2311 value: str # content of the comment
2312 newlines: int # how many newlines before the comment
2313 consumed: int # how many characters of the original leaf's prefix did we consume
2316 @lru_cache(maxsize=4096)
2317 def list_comments(prefix: str, *, is_endmarker: bool) -> List[ProtoComment]:
2318 """Return a list of :class:`ProtoComment` objects parsed from the given `prefix`."""
2319 result: List[ProtoComment] = []
2320 if not prefix or "#" not in prefix:
2326 for index, line in enumerate(prefix.split("\n")):
2327 consumed += len(line) + 1 # adding the length of the split '\n'
2328 line = line.lstrip()
2331 if not line.startswith("#"):
2332 # Escaped newlines outside of a comment are not really newlines at
2333 # all. We treat a single-line comment following an escaped newline
2334 # as a simple trailing comment.
2335 if line.endswith("\\"):
2339 if index == ignored_lines and not is_endmarker:
2340 comment_type = token.COMMENT # simple trailing comment
2342 comment_type = STANDALONE_COMMENT
2343 comment = make_comment(line)
2346 type=comment_type, value=comment, newlines=nlines, consumed=consumed
2353 def make_comment(content: str) -> str:
2354 """Return a consistently formatted comment from the given `content` string.
2356 All comments (except for "##", "#!", "#:", '#'", "#%%") should have a single
2357 space between the hash sign and the content.
2359 If `content` didn't start with a hash sign, one is provided.
2361 content = content.rstrip()
2365 if content[0] == "#":
2366 content = content[1:]
2367 if content and content[0] not in " !:#'%":
2368 content = " " + content
2369 return "#" + content
2375 inner: bool = False,
2376 features: Collection[Feature] = (),
2377 ) -> Iterator[Line]:
2378 """Split a `line` into potentially many lines.
2380 They should fit in the allotted `line_length` but might not be able to.
2381 `inner` signifies that there were a pair of brackets somewhere around the
2382 current `line`, possibly transitively. This means we can fallback to splitting
2383 by delimiters if the LHS/RHS don't yield any results.
2385 `features` are syntactical features that may be used in the output.
2391 line_str = str(line).strip("\n")
2394 not line.contains_uncollapsable_type_comments()
2395 and not line.should_explode
2396 and not line.is_collection_with_optional_trailing_comma
2398 is_line_short_enough(line, line_length=line_length, line_str=line_str)
2399 or line.contains_unsplittable_type_ignore()
2405 split_funcs: List[SplitFunc]
2407 split_funcs = [left_hand_split]
2410 def rhs(line: Line, features: Collection[Feature]) -> Iterator[Line]:
2411 for omit in generate_trailers_to_omit(line, line_length):
2412 lines = list(right_hand_split(line, line_length, features, omit=omit))
2413 if is_line_short_enough(lines[0], line_length=line_length):
2417 # All splits failed, best effort split with no omits.
2418 # This mostly happens to multiline strings that are by definition
2419 # reported as not fitting a single line.
2420 # line_length=1 here was historically a bug that somehow became a feature.
2421 # See #762 and #781 for the full story.
2422 yield from right_hand_split(line, line_length=1, features=features)
2424 if line.inside_brackets:
2425 split_funcs = [delimiter_split, standalone_comment_split, rhs]
2428 for split_func in split_funcs:
2429 # We are accumulating lines in `result` because we might want to abort
2430 # mission and return the original line in the end, or attempt a different
2432 result: List[Line] = []
2434 for l in split_func(line, features):
2435 if str(l).strip("\n") == line_str:
2436 raise CannotSplit("Split function returned an unchanged result")
2440 l, line_length=line_length, inner=True, features=features
2454 def left_hand_split(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2455 """Split line into many lines, starting with the first matching bracket pair.
2457 Note: this usually looks weird, only use this for function definitions.
2458 Prefer RHS otherwise. This is why this function is not symmetrical with
2459 :func:`right_hand_split` which also handles optional parentheses.
2461 tail_leaves: List[Leaf] = []
2462 body_leaves: List[Leaf] = []
2463 head_leaves: List[Leaf] = []
2464 current_leaves = head_leaves
2465 matching_bracket = None
2466 for leaf in line.leaves:
2468 current_leaves is body_leaves
2469 and leaf.type in CLOSING_BRACKETS
2470 and leaf.opening_bracket is matching_bracket
2472 current_leaves = tail_leaves if body_leaves else head_leaves
2473 current_leaves.append(leaf)
2474 if current_leaves is head_leaves:
2475 if leaf.type in OPENING_BRACKETS:
2476 matching_bracket = leaf
2477 current_leaves = body_leaves
2478 if not matching_bracket:
2479 raise CannotSplit("No brackets found")
2481 head = bracket_split_build_line(head_leaves, line, matching_bracket)
2482 body = bracket_split_build_line(body_leaves, line, matching_bracket, is_body=True)
2483 tail = bracket_split_build_line(tail_leaves, line, matching_bracket)
2484 bracket_split_succeeded_or_raise(head, body, tail)
2485 for result in (head, body, tail):
2490 def right_hand_split(
2493 features: Collection[Feature] = (),
2494 omit: Collection[LeafID] = (),
2495 ) -> Iterator[Line]:
2496 """Split line into many lines, starting with the last matching bracket pair.
2498 If the split was by optional parentheses, attempt splitting without them, too.
2499 `omit` is a collection of closing bracket IDs that shouldn't be considered for
2502 Note: running this function modifies `bracket_depth` on the leaves of `line`.
2504 tail_leaves: List[Leaf] = []
2505 body_leaves: List[Leaf] = []
2506 head_leaves: List[Leaf] = []
2507 current_leaves = tail_leaves
2508 opening_bracket = None
2509 closing_bracket = None
2510 for leaf in reversed(line.leaves):
2511 if current_leaves is body_leaves:
2512 if leaf is opening_bracket:
2513 current_leaves = head_leaves if body_leaves else tail_leaves
2514 current_leaves.append(leaf)
2515 if current_leaves is tail_leaves:
2516 if leaf.type in CLOSING_BRACKETS and id(leaf) not in omit:
2517 opening_bracket = leaf.opening_bracket
2518 closing_bracket = leaf
2519 current_leaves = body_leaves
2520 if not (opening_bracket and closing_bracket and head_leaves):
2521 # If there is no opening or closing_bracket that means the split failed and
2522 # all content is in the tail. Otherwise, if `head_leaves` are empty, it means
2523 # the matching `opening_bracket` wasn't available on `line` anymore.
2524 raise CannotSplit("No brackets found")
2526 tail_leaves.reverse()
2527 body_leaves.reverse()
2528 head_leaves.reverse()
2529 head = bracket_split_build_line(head_leaves, line, opening_bracket)
2530 body = bracket_split_build_line(body_leaves, line, opening_bracket, is_body=True)
2531 tail = bracket_split_build_line(tail_leaves, line, opening_bracket)
2532 bracket_split_succeeded_or_raise(head, body, tail)
2534 # the body shouldn't be exploded
2535 not body.should_explode
2536 # the opening bracket is an optional paren
2537 and opening_bracket.type == token.LPAR
2538 and not opening_bracket.value
2539 # the closing bracket is an optional paren
2540 and closing_bracket.type == token.RPAR
2541 and not closing_bracket.value
2542 # it's not an import (optional parens are the only thing we can split on
2543 # in this case; attempting a split without them is a waste of time)
2544 and not line.is_import
2545 # there are no standalone comments in the body
2546 and not body.contains_standalone_comments(0)
2547 # and we can actually remove the parens
2548 and can_omit_invisible_parens(body, line_length)
2550 omit = {id(closing_bracket), *omit}
2552 yield from right_hand_split(line, line_length, features=features, omit=omit)
2558 or is_line_short_enough(body, line_length=line_length)
2561 "Splitting failed, body is still too long and can't be split."
2564 elif head.contains_multiline_strings() or tail.contains_multiline_strings():
2566 "The current optional pair of parentheses is bound to fail to "
2567 "satisfy the splitting algorithm because the head or the tail "
2568 "contains multiline strings which by definition never fit one "
2572 ensure_visible(opening_bracket)
2573 ensure_visible(closing_bracket)
2574 for result in (head, body, tail):
2579 def bracket_split_succeeded_or_raise(head: Line, body: Line, tail: Line) -> None:
2580 """Raise :exc:`CannotSplit` if the last left- or right-hand split failed.
2582 Do nothing otherwise.
2584 A left- or right-hand split is based on a pair of brackets. Content before
2585 (and including) the opening bracket is left on one line, content inside the
2586 brackets is put on a separate line, and finally content starting with and
2587 following the closing bracket is put on a separate line.
2589 Those are called `head`, `body`, and `tail`, respectively. If the split
2590 produced the same line (all content in `head`) or ended up with an empty `body`
2591 and the `tail` is just the closing bracket, then it's considered failed.
2593 tail_len = len(str(tail).strip())
2596 raise CannotSplit("Splitting brackets produced the same line")
2600 f"Splitting brackets on an empty body to save "
2601 f"{tail_len} characters is not worth it"
2605 def bracket_split_build_line(
2606 leaves: List[Leaf], original: Line, opening_bracket: Leaf, *, is_body: bool = False
2608 """Return a new line with given `leaves` and respective comments from `original`.
2610 If `is_body` is True, the result line is one-indented inside brackets and as such
2611 has its first leaf's prefix normalized and a trailing comma added when expected.
2613 result = Line(depth=original.depth)
2615 result.inside_brackets = True
2618 # Since body is a new indent level, remove spurious leading whitespace.
2619 normalize_prefix(leaves[0], inside_brackets=True)
2620 # Ensure a trailing comma for imports and standalone function arguments, but
2621 # be careful not to add one after any comments or within type annotations.
2624 and opening_bracket.value == "("
2625 and not any(l.type == token.COMMA for l in leaves)
2628 if original.is_import or no_commas:
2629 for i in range(len(leaves) - 1, -1, -1):
2630 if leaves[i].type == STANDALONE_COMMENT:
2633 if leaves[i].type != token.COMMA:
2634 leaves.insert(i + 1, Leaf(token.COMMA, ","))
2639 result.append(leaf, preformatted=True)
2640 for comment_after in original.comments_after(leaf):
2641 result.append(comment_after, preformatted=True)
2643 result.should_explode = should_explode(result, opening_bracket)
2647 def dont_increase_indentation(split_func: SplitFunc) -> SplitFunc:
2648 """Normalize prefix of the first leaf in every line returned by `split_func`.
2650 This is a decorator over relevant split functions.
2654 def split_wrapper(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2655 for l in split_func(line, features):
2656 normalize_prefix(l.leaves[0], inside_brackets=True)
2659 return split_wrapper
2662 @dont_increase_indentation
2663 def delimiter_split(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2664 """Split according to delimiters of the highest priority.
2666 If the appropriate Features are given, the split will add trailing commas
2667 also in function signatures and calls that contain `*` and `**`.
2670 last_leaf = line.leaves[-1]
2672 raise CannotSplit("Line empty")
2674 bt = line.bracket_tracker
2676 delimiter_priority = bt.max_delimiter_priority(exclude={id(last_leaf)})
2678 raise CannotSplit("No delimiters found")
2680 if delimiter_priority == DOT_PRIORITY:
2681 if bt.delimiter_count_with_priority(delimiter_priority) == 1:
2682 raise CannotSplit("Splitting a single attribute from its owner looks wrong")
2684 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2685 lowest_depth = sys.maxsize
2686 trailing_comma_safe = True
2688 def append_to_line(leaf: Leaf) -> Iterator[Line]:
2689 """Append `leaf` to current line or to new line if appending impossible."""
2690 nonlocal current_line
2692 current_line.append_safe(leaf, preformatted=True)
2696 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2697 current_line.append(leaf)
2699 for leaf in line.leaves:
2700 yield from append_to_line(leaf)
2702 for comment_after in line.comments_after(leaf):
2703 yield from append_to_line(comment_after)
2705 lowest_depth = min(lowest_depth, leaf.bracket_depth)
2706 if leaf.bracket_depth == lowest_depth:
2707 if is_vararg(leaf, within={syms.typedargslist}):
2708 trailing_comma_safe = (
2709 trailing_comma_safe and Feature.TRAILING_COMMA_IN_DEF in features
2711 elif is_vararg(leaf, within={syms.arglist, syms.argument}):
2712 trailing_comma_safe = (
2713 trailing_comma_safe and Feature.TRAILING_COMMA_IN_CALL in features
2716 leaf_priority = bt.delimiters.get(id(leaf))
2717 if leaf_priority == delimiter_priority:
2720 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2724 and delimiter_priority == COMMA_PRIORITY
2725 and current_line.leaves[-1].type != token.COMMA
2726 and current_line.leaves[-1].type != STANDALONE_COMMENT
2728 current_line.append(Leaf(token.COMMA, ","))
2732 @dont_increase_indentation
2733 def standalone_comment_split(
2734 line: Line, features: Collection[Feature] = ()
2735 ) -> Iterator[Line]:
2736 """Split standalone comments from the rest of the line."""
2737 if not line.contains_standalone_comments(0):
2738 raise CannotSplit("Line does not have any standalone comments")
2740 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2742 def append_to_line(leaf: Leaf) -> Iterator[Line]:
2743 """Append `leaf` to current line or to new line if appending impossible."""
2744 nonlocal current_line
2746 current_line.append_safe(leaf, preformatted=True)
2750 current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2751 current_line.append(leaf)
2753 for leaf in line.leaves:
2754 yield from append_to_line(leaf)
2756 for comment_after in line.comments_after(leaf):
2757 yield from append_to_line(comment_after)
2763 def is_import(leaf: Leaf) -> bool:
2764 """Return True if the given leaf starts an import statement."""
2771 (v == "import" and p and p.type == syms.import_name)
2772 or (v == "from" and p and p.type == syms.import_from)
2777 def is_type_comment(leaf: Leaf, suffix: str = "") -> bool:
2778 """Return True if the given leaf is a special comment.
2779 Only returns true for type comments for now."""
2782 return t in {token.COMMENT, STANDALONE_COMMENT} and v.startswith("# type:" + suffix)
2785 def normalize_prefix(leaf: Leaf, *, inside_brackets: bool) -> None:
2786 """Leave existing extra newlines if not `inside_brackets`. Remove everything
2789 Note: don't use backslashes for formatting or you'll lose your voting rights.
2791 if not inside_brackets:
2792 spl = leaf.prefix.split("#")
2793 if "\\" not in spl[0]:
2794 nl_count = spl[-1].count("\n")
2797 leaf.prefix = "\n" * nl_count
2803 def normalize_string_prefix(leaf: Leaf, remove_u_prefix: bool = False) -> None:
2804 """Make all string prefixes lowercase.
2806 If remove_u_prefix is given, also removes any u prefix from the string.
2808 Note: Mutates its argument.
2810 match = re.match(r"^([furbFURB]*)(.*)$", leaf.value, re.DOTALL)
2811 assert match is not None, f"failed to match string {leaf.value!r}"
2812 orig_prefix = match.group(1)
2813 new_prefix = orig_prefix.lower()
2815 new_prefix = new_prefix.replace("u", "")
2816 leaf.value = f"{new_prefix}{match.group(2)}"
2819 def normalize_string_quotes(leaf: Leaf) -> None:
2820 """Prefer double quotes but only if it doesn't cause more escaping.
2822 Adds or removes backslashes as appropriate. Doesn't parse and fix
2823 strings nested in f-strings (yet).
2825 Note: Mutates its argument.
2827 value = leaf.value.lstrip("furbFURB")
2828 if value[:3] == '"""':
2831 elif value[:3] == "'''":
2834 elif value[0] == '"':
2840 first_quote_pos = leaf.value.find(orig_quote)
2841 if first_quote_pos == -1:
2842 return # There's an internal error
2844 prefix = leaf.value[:first_quote_pos]
2845 unescaped_new_quote = re.compile(rf"(([^\\]|^)(\\\\)*){new_quote}")
2846 escaped_new_quote = re.compile(rf"([^\\]|^)\\((?:\\\\)*){new_quote}")
2847 escaped_orig_quote = re.compile(rf"([^\\]|^)\\((?:\\\\)*){orig_quote}")
2848 body = leaf.value[first_quote_pos + len(orig_quote) : -len(orig_quote)]
2849 if "r" in prefix.casefold():
2850 if unescaped_new_quote.search(body):
2851 # There's at least one unescaped new_quote in this raw string
2852 # so converting is impossible
2855 # Do not introduce or remove backslashes in raw strings
2858 # remove unnecessary escapes
2859 new_body = sub_twice(escaped_new_quote, rf"\1\2{new_quote}", body)
2860 if body != new_body:
2861 # Consider the string without unnecessary escapes as the original
2863 leaf.value = f"{prefix}{orig_quote}{body}{orig_quote}"
2864 new_body = sub_twice(escaped_orig_quote, rf"\1\2{orig_quote}", new_body)
2865 new_body = sub_twice(unescaped_new_quote, rf"\1\\{new_quote}", new_body)
2866 if "f" in prefix.casefold():
2867 matches = re.findall(
2869 (?:[^{]|^)\{ # start of the string or a non-{ followed by a single {
2870 ([^{].*?) # contents of the brackets except if begins with {{
2871 \}(?:[^}]|$) # A } followed by end of the string or a non-}
2878 # Do not introduce backslashes in interpolated expressions
2881 if new_quote == '"""' and new_body[-1:] == '"':
2883 new_body = new_body[:-1] + '\\"'
2884 orig_escape_count = body.count("\\")
2885 new_escape_count = new_body.count("\\")
2886 if new_escape_count > orig_escape_count:
2887 return # Do not introduce more escaping
2889 if new_escape_count == orig_escape_count and orig_quote == '"':
2890 return # Prefer double quotes
2892 leaf.value = f"{prefix}{new_quote}{new_body}{new_quote}"
2895 def normalize_numeric_literal(leaf: Leaf) -> None:
2896 """Normalizes numeric (float, int, and complex) literals.
2898 All letters used in the representation are normalized to lowercase (except
2899 in Python 2 long literals).
2901 text = leaf.value.lower()
2902 if text.startswith(("0o", "0b")):
2903 # Leave octal and binary literals alone.
2905 elif text.startswith("0x"):
2906 # Change hex literals to upper case.
2907 before, after = text[:2], text[2:]
2908 text = f"{before}{after.upper()}"
2910 before, after = text.split("e")
2912 if after.startswith("-"):
2915 elif after.startswith("+"):
2917 before = format_float_or_int_string(before)
2918 text = f"{before}e{sign}{after}"
2919 elif text.endswith(("j", "l")):
2922 # Capitalize in "2L" because "l" looks too similar to "1".
2925 text = f"{format_float_or_int_string(number)}{suffix}"
2927 text = format_float_or_int_string(text)
2931 def format_float_or_int_string(text: str) -> str:
2932 """Formats a float string like "1.0"."""
2936 before, after = text.split(".")
2937 return f"{before or 0}.{after or 0}"
2940 def normalize_invisible_parens(node: Node, parens_after: Set[str]) -> None:
2941 """Make existing optional parentheses invisible or create new ones.
2943 `parens_after` is a set of string leaf values immediately after which parens
2946 Standardizes on visible parentheses for single-element tuples, and keeps
2947 existing visible parentheses for other tuples and generator expressions.
2949 for pc in list_comments(node.prefix, is_endmarker=False):
2950 if pc.value in FMT_OFF:
2951 # This `node` has a prefix with `# fmt: off`, don't mess with parens.
2954 for index, child in enumerate(list(node.children)):
2955 # Add parentheses around long tuple unpacking in assignments.
2958 and isinstance(child, Node)
2959 and child.type == syms.testlist_star_expr
2964 if is_walrus_assignment(child):
2967 if child.type == syms.atom:
2968 if maybe_make_parens_invisible_in_atom(child, parent=node):
2969 wrap_in_parentheses(node, child, visible=False)
2970 elif is_one_tuple(child):
2971 wrap_in_parentheses(node, child, visible=True)
2972 elif node.type == syms.import_from:
2973 # "import from" nodes store parentheses directly as part of
2975 if child.type == token.LPAR:
2976 # make parentheses invisible
2977 child.value = "" # type: ignore
2978 node.children[-1].value = "" # type: ignore
2979 elif child.type != token.STAR:
2980 # insert invisible parentheses
2981 node.insert_child(index, Leaf(token.LPAR, ""))
2982 node.append_child(Leaf(token.RPAR, ""))
2985 elif not (isinstance(child, Leaf) and is_multiline_string(child)):
2986 wrap_in_parentheses(node, child, visible=False)
2988 check_lpar = isinstance(child, Leaf) and child.value in parens_after
2991 def normalize_fmt_off(node: Node) -> None:
2992 """Convert content between `# fmt: off`/`# fmt: on` into standalone comments."""
2995 try_again = convert_one_fmt_off_pair(node)
2998 def convert_one_fmt_off_pair(node: Node) -> bool:
2999 """Convert content of a single `# fmt: off`/`# fmt: on` into a standalone comment.
3001 Returns True if a pair was converted.
3003 for leaf in node.leaves():
3004 previous_consumed = 0
3005 for comment in list_comments(leaf.prefix, is_endmarker=False):
3006 if comment.value in FMT_OFF:
3007 # We only want standalone comments. If there's no previous leaf or
3008 # the previous leaf is indentation, it's a standalone comment in
3010 if comment.type != STANDALONE_COMMENT:
3011 prev = preceding_leaf(leaf)
3012 if prev and prev.type not in WHITESPACE:
3015 ignored_nodes = list(generate_ignored_nodes(leaf))
3016 if not ignored_nodes:
3019 first = ignored_nodes[0] # Can be a container node with the `leaf`.
3020 parent = first.parent
3021 prefix = first.prefix
3022 first.prefix = prefix[comment.consumed :]
3024 comment.value + "\n" + "".join(str(n) for n in ignored_nodes)
3026 if hidden_value.endswith("\n"):
3027 # That happens when one of the `ignored_nodes` ended with a NEWLINE
3028 # leaf (possibly followed by a DEDENT).
3029 hidden_value = hidden_value[:-1]
3031 for ignored in ignored_nodes:
3032 index = ignored.remove()
3033 if first_idx is None:
3035 assert parent is not None, "INTERNAL ERROR: fmt: on/off handling (1)"
3036 assert first_idx is not None, "INTERNAL ERROR: fmt: on/off handling (2)"
3037 parent.insert_child(
3042 prefix=prefix[:previous_consumed] + "\n" * comment.newlines,
3047 previous_consumed = comment.consumed
3052 def generate_ignored_nodes(leaf: Leaf) -> Iterator[LN]:
3053 """Starting from the container of `leaf`, generate all leaves until `# fmt: on`.
3055 Stops at the end of the block.
3057 container: Optional[LN] = container_of(leaf)
3058 while container is not None and container.type != token.ENDMARKER:
3060 for comment in list_comments(container.prefix, is_endmarker=False):
3061 if comment.value in FMT_ON:
3063 elif comment.value in FMT_OFF:
3070 container = container.next_sibling
3073 def maybe_make_parens_invisible_in_atom(node: LN, parent: LN) -> bool:
3074 """If it's safe, make the parens in the atom `node` invisible, recursively.
3075 Additionally, remove repeated, adjacent invisible parens from the atom `node`
3076 as they are redundant.
3078 Returns whether the node should itself be wrapped in invisible parentheses.
3082 node.type != syms.atom
3083 or is_empty_tuple(node)
3084 or is_one_tuple(node)
3085 or (is_yield(node) and parent.type != syms.expr_stmt)
3086 or max_delimiter_priority_in_atom(node) >= COMMA_PRIORITY
3090 first = node.children[0]
3091 last = node.children[-1]
3092 if first.type == token.LPAR and last.type == token.RPAR:
3093 middle = node.children[1]
3094 # make parentheses invisible
3095 first.value = "" # type: ignore
3096 last.value = "" # type: ignore
3097 maybe_make_parens_invisible_in_atom(middle, parent=parent)
3099 if is_atom_with_invisible_parens(middle):
3100 # Strip the invisible parens from `middle` by replacing
3101 # it with the child in-between the invisible parens
3102 middle.replace(middle.children[1])
3109 def is_atom_with_invisible_parens(node: LN) -> bool:
3110 """Given a `LN`, determines whether it's an atom `node` with invisible
3111 parens. Useful in dedupe-ing and normalizing parens.
3113 if isinstance(node, Leaf) or node.type != syms.atom:
3116 first, last = node.children[0], node.children[-1]
3118 isinstance(first, Leaf)
3119 and first.type == token.LPAR
3120 and first.value == ""
3121 and isinstance(last, Leaf)
3122 and last.type == token.RPAR
3123 and last.value == ""
3127 def is_empty_tuple(node: LN) -> bool:
3128 """Return True if `node` holds an empty tuple."""
3130 node.type == syms.atom
3131 and len(node.children) == 2
3132 and node.children[0].type == token.LPAR
3133 and node.children[1].type == token.RPAR
3137 def unwrap_singleton_parenthesis(node: LN) -> Optional[LN]:
3138 """Returns `wrapped` if `node` is of the shape ( wrapped ).
3140 Parenthesis can be optional. Returns None otherwise"""
3141 if len(node.children) != 3:
3144 lpar, wrapped, rpar = node.children
3145 if not (lpar.type == token.LPAR and rpar.type == token.RPAR):
3151 def wrap_in_parentheses(parent: Node, child: LN, *, visible: bool = True) -> None:
3152 """Wrap `child` in parentheses.
3154 This replaces `child` with an atom holding the parentheses and the old
3155 child. That requires moving the prefix.
3157 If `visible` is False, the leaves will be valueless (and thus invisible).
3159 lpar = Leaf(token.LPAR, "(" if visible else "")
3160 rpar = Leaf(token.RPAR, ")" if visible else "")
3161 prefix = child.prefix
3163 index = child.remove() or 0
3164 new_child = Node(syms.atom, [lpar, child, rpar])
3165 new_child.prefix = prefix
3166 parent.insert_child(index, new_child)
3169 def is_one_tuple(node: LN) -> bool:
3170 """Return True if `node` holds a tuple with one element, with or without parens."""
3171 if node.type == syms.atom:
3172 gexp = unwrap_singleton_parenthesis(node)
3173 if gexp is None or gexp.type != syms.testlist_gexp:
3176 return len(gexp.children) == 2 and gexp.children[1].type == token.COMMA
3179 node.type in IMPLICIT_TUPLE
3180 and len(node.children) == 2
3181 and node.children[1].type == token.COMMA
3185 def is_walrus_assignment(node: LN) -> bool:
3186 """Return True iff `node` is of the shape ( test := test )"""
3187 inner = unwrap_singleton_parenthesis(node)
3188 return inner is not None and inner.type == syms.namedexpr_test
3191 def is_yield(node: LN) -> bool:
3192 """Return True if `node` holds a `yield` or `yield from` expression."""
3193 if node.type == syms.yield_expr:
3196 if node.type == token.NAME and node.value == "yield": # type: ignore
3199 if node.type != syms.atom:
3202 if len(node.children) != 3:
3205 lpar, expr, rpar = node.children
3206 if lpar.type == token.LPAR and rpar.type == token.RPAR:
3207 return is_yield(expr)
3212 def is_vararg(leaf: Leaf, within: Set[NodeType]) -> bool:
3213 """Return True if `leaf` is a star or double star in a vararg or kwarg.
3215 If `within` includes VARARGS_PARENTS, this applies to function signatures.
3216 If `within` includes UNPACKING_PARENTS, it applies to right hand-side
3217 extended iterable unpacking (PEP 3132) and additional unpacking
3218 generalizations (PEP 448).
3220 if leaf.type not in VARARGS_SPECIALS or not leaf.parent:
3224 if p.type == syms.star_expr:
3225 # Star expressions are also used as assignment targets in extended
3226 # iterable unpacking (PEP 3132). See what its parent is instead.
3232 return p.type in within
3235 def is_multiline_string(leaf: Leaf) -> bool:
3236 """Return True if `leaf` is a multiline string that actually spans many lines."""
3237 value = leaf.value.lstrip("furbFURB")
3238 return value[:3] in {'"""', "'''"} and "\n" in value
3241 def is_stub_suite(node: Node) -> bool:
3242 """Return True if `node` is a suite with a stub body."""
3244 len(node.children) != 4
3245 or node.children[0].type != token.NEWLINE
3246 or node.children[1].type != token.INDENT
3247 or node.children[3].type != token.DEDENT
3251 return is_stub_body(node.children[2])
3254 def is_stub_body(node: LN) -> bool:
3255 """Return True if `node` is a simple statement containing an ellipsis."""
3256 if not isinstance(node, Node) or node.type != syms.simple_stmt:
3259 if len(node.children) != 2:
3262 child = node.children[0]
3264 child.type == syms.atom
3265 and len(child.children) == 3
3266 and all(leaf == Leaf(token.DOT, ".") for leaf in child.children)
3270 def max_delimiter_priority_in_atom(node: LN) -> Priority:
3271 """Return maximum delimiter priority inside `node`.
3273 This is specific to atoms with contents contained in a pair of parentheses.
3274 If `node` isn't an atom or there are no enclosing parentheses, returns 0.
3276 if node.type != syms.atom:
3279 first = node.children[0]
3280 last = node.children[-1]
3281 if not (first.type == token.LPAR and last.type == token.RPAR):
3284 bt = BracketTracker()
3285 for c in node.children[1:-1]:
3286 if isinstance(c, Leaf):
3289 for leaf in c.leaves():
3292 return bt.max_delimiter_priority()
3298 def ensure_visible(leaf: Leaf) -> None:
3299 """Make sure parentheses are visible.
3301 They could be invisible as part of some statements (see
3302 :func:`normalize_invisible_parens` and :func:`visit_import_from`).
3304 if leaf.type == token.LPAR:
3306 elif leaf.type == token.RPAR:
3310 def should_explode(line: Line, opening_bracket: Leaf) -> bool:
3311 """Should `line` immediately be split with `delimiter_split()` after RHS?"""
3314 opening_bracket.parent
3315 and opening_bracket.parent.type in {syms.atom, syms.import_from}
3316 and opening_bracket.value in "[{("
3321 last_leaf = line.leaves[-1]
3322 exclude = {id(last_leaf)} if last_leaf.type == token.COMMA else set()
3323 max_priority = line.bracket_tracker.max_delimiter_priority(exclude=exclude)
3324 except (IndexError, ValueError):
3327 return max_priority == COMMA_PRIORITY
3330 def get_features_used(node: Node) -> Set[Feature]:
3331 """Return a set of (relatively) new Python features used in this file.
3333 Currently looking for:
3335 - underscores in numeric literals;
3336 - trailing commas after * or ** in function signatures and calls;
3337 - positional only arguments in function signatures and lambdas;
3339 features: Set[Feature] = set()
3340 for n in node.pre_order():
3341 if n.type == token.STRING:
3342 value_head = n.value[:2] # type: ignore
3343 if value_head in {'f"', 'F"', "f'", "F'", "rf", "fr", "RF", "FR"}:
3344 features.add(Feature.F_STRINGS)
3346 elif n.type == token.NUMBER:
3347 if "_" in n.value: # type: ignore
3348 features.add(Feature.NUMERIC_UNDERSCORES)
3350 elif n.type == token.SLASH:
3351 if n.parent and n.parent.type in {syms.typedargslist, syms.arglist}:
3352 features.add(Feature.POS_ONLY_ARGUMENTS)
3354 elif n.type == token.COLONEQUAL:
3355 features.add(Feature.ASSIGNMENT_EXPRESSIONS)
3358 n.type in {syms.typedargslist, syms.arglist}
3360 and n.children[-1].type == token.COMMA
3362 if n.type == syms.typedargslist:
3363 feature = Feature.TRAILING_COMMA_IN_DEF
3365 feature = Feature.TRAILING_COMMA_IN_CALL
3367 for ch in n.children:
3368 if ch.type in STARS:
3369 features.add(feature)
3371 if ch.type == syms.argument:
3372 for argch in ch.children:
3373 if argch.type in STARS:
3374 features.add(feature)
3379 def detect_target_versions(node: Node) -> Set[TargetVersion]:
3380 """Detect the version to target based on the nodes used."""
3381 features = get_features_used(node)
3383 version for version in TargetVersion if features <= VERSION_TO_FEATURES[version]
3387 def generate_trailers_to_omit(line: Line, line_length: int) -> Iterator[Set[LeafID]]:
3388 """Generate sets of closing bracket IDs that should be omitted in a RHS.
3390 Brackets can be omitted if the entire trailer up to and including
3391 a preceding closing bracket fits in one line.
3393 Yielded sets are cumulative (contain results of previous yields, too). First
3397 omit: Set[LeafID] = set()
3400 length = 4 * line.depth
3401 opening_bracket = None
3402 closing_bracket = None
3403 inner_brackets: Set[LeafID] = set()
3404 for index, leaf, leaf_length in enumerate_with_length(line, reversed=True):
3405 length += leaf_length
3406 if length > line_length:
3409 has_inline_comment = leaf_length > len(leaf.value) + len(leaf.prefix)
3410 if leaf.type == STANDALONE_COMMENT or has_inline_comment:
3414 if leaf is opening_bracket:
3415 opening_bracket = None
3416 elif leaf.type in CLOSING_BRACKETS:
3417 inner_brackets.add(id(leaf))
3418 elif leaf.type in CLOSING_BRACKETS:
3419 if index > 0 and line.leaves[index - 1].type in OPENING_BRACKETS:
3420 # Empty brackets would fail a split so treat them as "inner"
3421 # brackets (e.g. only add them to the `omit` set if another
3422 # pair of brackets was good enough.
3423 inner_brackets.add(id(leaf))
3427 omit.add(id(closing_bracket))
3428 omit.update(inner_brackets)
3429 inner_brackets.clear()
3433 opening_bracket = leaf.opening_bracket
3434 closing_bracket = leaf
3437 def get_future_imports(node: Node) -> Set[str]:
3438 """Return a set of __future__ imports in the file."""
3439 imports: Set[str] = set()
3441 def get_imports_from_children(children: List[LN]) -> Generator[str, None, None]:
3442 for child in children:
3443 if isinstance(child, Leaf):
3444 if child.type == token.NAME:
3447 elif child.type == syms.import_as_name:
3448 orig_name = child.children[0]
3449 assert isinstance(orig_name, Leaf), "Invalid syntax parsing imports"
3450 assert orig_name.type == token.NAME, "Invalid syntax parsing imports"
3451 yield orig_name.value
3453 elif child.type == syms.import_as_names:
3454 yield from get_imports_from_children(child.children)
3457 raise AssertionError("Invalid syntax parsing imports")
3459 for child in node.children:
3460 if child.type != syms.simple_stmt:
3463 first_child = child.children[0]
3464 if isinstance(first_child, Leaf):
3465 # Continue looking if we see a docstring; otherwise stop.
3467 len(child.children) == 2
3468 and first_child.type == token.STRING
3469 and child.children[1].type == token.NEWLINE
3475 elif first_child.type == syms.import_from:
3476 module_name = first_child.children[1]
3477 if not isinstance(module_name, Leaf) or module_name.value != "__future__":
3480 imports |= set(get_imports_from_children(first_child.children[3:]))
3488 def get_gitignore(root: Path) -> PathSpec:
3489 """ Return a PathSpec matching gitignore content if present."""
3490 gitignore = root / ".gitignore"
3491 lines: List[str] = []
3492 if gitignore.is_file():
3493 with gitignore.open() as gf:
3494 lines = gf.readlines()
3495 return PathSpec.from_lines("gitwildmatch", lines)
3498 def gen_python_files_in_dir(
3501 include: Pattern[str],
3502 exclude: Pattern[str],
3504 gitignore: PathSpec,
3505 ) -> Iterator[Path]:
3506 """Generate all files under `path` whose paths are not excluded by the
3507 `exclude` regex, but are included by the `include` regex.
3509 Symbolic links pointing outside of the `root` directory are ignored.
3511 `report` is where output about exclusions goes.
3513 assert root.is_absolute(), f"INTERNAL ERROR: `root` must be absolute but is {root}"
3514 for child in path.iterdir():
3515 # First ignore files matching .gitignore
3516 if gitignore.match_file(child.as_posix()):
3517 report.path_ignored(child, f"matches the .gitignore file content")
3520 # Then ignore with `exclude` option.
3522 normalized_path = "/" + child.resolve().relative_to(root).as_posix()
3523 except OSError as e:
3524 report.path_ignored(child, f"cannot be read because {e}")
3528 if child.is_symlink():
3529 report.path_ignored(
3530 child, f"is a symbolic link that points outside {root}"
3537 normalized_path += "/"
3539 exclude_match = exclude.search(normalized_path)
3540 if exclude_match and exclude_match.group(0):
3541 report.path_ignored(child, f"matches the --exclude regular expression")
3545 yield from gen_python_files_in_dir(
3546 child, root, include, exclude, report, gitignore
3549 elif child.is_file():
3550 include_match = include.search(normalized_path)
3556 def find_project_root(srcs: Iterable[str]) -> Path:
3557 """Return a directory containing .git, .hg, or pyproject.toml.
3559 That directory can be one of the directories passed in `srcs` or their
3562 If no directory in the tree contains a marker that would specify it's the
3563 project root, the root of the file system is returned.
3566 return Path("/").resolve()
3568 common_base = min(Path(src).resolve() for src in srcs)
3569 if common_base.is_dir():
3570 # Append a fake file so `parents` below returns `common_base_dir`, too.
3571 common_base /= "fake-file"
3572 for directory in common_base.parents:
3573 if (directory / ".git").is_dir():
3576 if (directory / ".hg").is_dir():
3579 if (directory / "pyproject.toml").is_file():
3587 """Provides a reformatting counter. Can be rendered with `str(report)`."""
3591 verbose: bool = False
3592 change_count: int = 0
3594 failure_count: int = 0
3596 def done(self, src: Path, changed: Changed) -> None:
3597 """Increment the counter for successful reformatting. Write out a message."""
3598 if changed is Changed.YES:
3599 reformatted = "would reformat" if self.check else "reformatted"
3600 if self.verbose or not self.quiet:
3601 out(f"{reformatted} {src}")
3602 self.change_count += 1
3605 if changed is Changed.NO:
3606 msg = f"{src} already well formatted, good job."
3608 msg = f"{src} wasn't modified on disk since last run."
3609 out(msg, bold=False)
3610 self.same_count += 1
3612 def failed(self, src: Path, message: str) -> None:
3613 """Increment the counter for failed reformatting. Write out a message."""
3614 err(f"error: cannot format {src}: {message}")
3615 self.failure_count += 1
3617 def path_ignored(self, path: Path, message: str) -> None:
3619 out(f"{path} ignored: {message}", bold=False)
3622 def return_code(self) -> int:
3623 """Return the exit code that the app should use.
3625 This considers the current state of changed files and failures:
3626 - if there were any failures, return 123;
3627 - if any files were changed and --check is being used, return 1;
3628 - otherwise return 0.
3630 # According to http://tldp.org/LDP/abs/html/exitcodes.html starting with
3631 # 126 we have special return codes reserved by the shell.
3632 if self.failure_count:
3635 elif self.change_count and self.check:
3640 def __str__(self) -> str:
3641 """Render a color report of the current state.
3643 Use `click.unstyle` to remove colors.
3646 reformatted = "would be reformatted"
3647 unchanged = "would be left unchanged"
3648 failed = "would fail to reformat"
3650 reformatted = "reformatted"
3651 unchanged = "left unchanged"
3652 failed = "failed to reformat"
3654 if self.change_count:
3655 s = "s" if self.change_count > 1 else ""
3657 click.style(f"{self.change_count} file{s} {reformatted}", bold=True)
3660 s = "s" if self.same_count > 1 else ""
3661 report.append(f"{self.same_count} file{s} {unchanged}")
3662 if self.failure_count:
3663 s = "s" if self.failure_count > 1 else ""
3665 click.style(f"{self.failure_count} file{s} {failed}", fg="red")
3667 return ", ".join(report) + "."
3670 def parse_ast(src: str) -> Union[ast.AST, ast3.AST, ast27.AST]:
3671 filename = "<unknown>"
3672 if sys.version_info >= (3, 8):
3673 # TODO: support Python 4+ ;)
3674 for minor_version in range(sys.version_info[1], 4, -1):
3676 return ast.parse(src, filename, feature_version=(3, minor_version))
3680 for feature_version in (7, 6):
3682 return ast3.parse(src, filename, feature_version=feature_version)
3686 return ast27.parse(src)
3689 def _fixup_ast_constants(
3690 node: Union[ast.AST, ast3.AST, ast27.AST]
3691 ) -> Union[ast.AST, ast3.AST, ast27.AST]:
3692 """Map ast nodes deprecated in 3.8 to Constant."""
3693 if isinstance(node, (ast.Str, ast3.Str, ast27.Str, ast.Bytes, ast3.Bytes)):
3694 return ast.Constant(value=node.s)
3696 if isinstance(node, (ast.Num, ast3.Num, ast27.Num)):
3697 return ast.Constant(value=node.n)
3699 if isinstance(node, (ast.NameConstant, ast3.NameConstant)):
3700 return ast.Constant(value=node.value)
3705 def assert_equivalent(src: str, dst: str) -> None:
3706 """Raise AssertionError if `src` and `dst` aren't equivalent."""
3708 def _v(node: Union[ast.AST, ast3.AST, ast27.AST], depth: int = 0) -> Iterator[str]:
3709 """Simple visitor generating strings to compare ASTs by content."""
3711 node = _fixup_ast_constants(node)
3713 yield f"{' ' * depth}{node.__class__.__name__}("
3715 for field in sorted(node._fields):
3716 # TypeIgnore has only one field 'lineno' which breaks this comparison
3717 type_ignore_classes = (ast3.TypeIgnore, ast27.TypeIgnore)
3718 if sys.version_info >= (3, 8):
3719 type_ignore_classes += (ast.TypeIgnore,)
3720 if isinstance(node, type_ignore_classes):
3724 value = getattr(node, field)
3725 except AttributeError:
3728 yield f"{' ' * (depth+1)}{field}="
3730 if isinstance(value, list):
3732 # Ignore nested tuples within del statements, because we may insert
3733 # parentheses and they change the AST.
3736 and isinstance(node, (ast.Delete, ast3.Delete, ast27.Delete))
3737 and isinstance(item, (ast.Tuple, ast3.Tuple, ast27.Tuple))
3739 for item in item.elts:
3740 yield from _v(item, depth + 2)
3742 elif isinstance(item, (ast.AST, ast3.AST, ast27.AST)):
3743 yield from _v(item, depth + 2)
3745 elif isinstance(value, (ast.AST, ast3.AST, ast27.AST)):
3746 yield from _v(value, depth + 2)
3749 yield f"{' ' * (depth+2)}{value!r}, # {value.__class__.__name__}"
3751 yield f"{' ' * depth}) # /{node.__class__.__name__}"
3754 src_ast = parse_ast(src)
3755 except Exception as exc:
3756 raise AssertionError(
3757 f"cannot use --safe with this file; failed to parse source file. "
3758 f"AST error message: {exc}"
3762 dst_ast = parse_ast(dst)
3763 except Exception as exc:
3764 log = dump_to_file("".join(traceback.format_tb(exc.__traceback__)), dst)
3765 raise AssertionError(
3766 f"INTERNAL ERROR: Black produced invalid code: {exc}. "
3767 f"Please report a bug on https://github.com/psf/black/issues. "
3768 f"This invalid output might be helpful: {log}"
3771 src_ast_str = "\n".join(_v(src_ast))
3772 dst_ast_str = "\n".join(_v(dst_ast))
3773 if src_ast_str != dst_ast_str:
3774 log = dump_to_file(diff(src_ast_str, dst_ast_str, "src", "dst"))
3775 raise AssertionError(
3776 f"INTERNAL ERROR: Black produced code that is not equivalent to "
3778 f"Please report a bug on https://github.com/psf/black/issues. "
3779 f"This diff might be helpful: {log}"
3783 def assert_stable(src: str, dst: str, mode: FileMode) -> None:
3784 """Raise AssertionError if `dst` reformats differently the second time."""
3785 newdst = format_str(dst, mode=mode)
3788 diff(src, dst, "source", "first pass"),
3789 diff(dst, newdst, "first pass", "second pass"),
3791 raise AssertionError(
3792 f"INTERNAL ERROR: Black produced different code on the second pass "
3793 f"of the formatter. "
3794 f"Please report a bug on https://github.com/psf/black/issues. "
3795 f"This diff might be helpful: {log}"
3799 def dump_to_file(*output: str) -> str:
3800 """Dump `output` to a temporary file. Return path to the file."""
3801 with tempfile.NamedTemporaryFile(
3802 mode="w", prefix="blk_", suffix=".log", delete=False, encoding="utf8"
3804 for lines in output:
3806 if lines and lines[-1] != "\n":
3812 def nullcontext() -> Iterator[None]:
3813 """Return an empty context manager.
3815 To be used like `nullcontext` in Python 3.7.
3820 def diff(a: str, b: str, a_name: str, b_name: str) -> str:
3821 """Return a unified diff string between strings `a` and `b`."""
3824 a_lines = [line + "\n" for line in a.split("\n")]
3825 b_lines = [line + "\n" for line in b.split("\n")]
3827 difflib.unified_diff(a_lines, b_lines, fromfile=a_name, tofile=b_name, n=5)
3831 def cancel(tasks: Iterable[asyncio.Task]) -> None:
3832 """asyncio signal handler that cancels all `tasks` and reports to stderr."""
3838 def shutdown(loop: asyncio.AbstractEventLoop) -> None:
3839 """Cancel all pending tasks on `loop`, wait for them, and close the loop."""
3841 if sys.version_info[:2] >= (3, 7):
3842 all_tasks = asyncio.all_tasks
3844 all_tasks = asyncio.Task.all_tasks
3845 # This part is borrowed from asyncio/runners.py in Python 3.7b2.
3846 to_cancel = [task for task in all_tasks(loop) if not task.done()]
3850 for task in to_cancel:
3852 loop.run_until_complete(
3853 asyncio.gather(*to_cancel, loop=loop, return_exceptions=True)
3856 # `concurrent.futures.Future` objects cannot be cancelled once they
3857 # are already running. There might be some when the `shutdown()` happened.
3858 # Silence their logger's spew about the event loop being closed.
3859 cf_logger = logging.getLogger("concurrent.futures")
3860 cf_logger.setLevel(logging.CRITICAL)
3864 def sub_twice(regex: Pattern[str], replacement: str, original: str) -> str:
3865 """Replace `regex` with `replacement` twice on `original`.
3867 This is used by string normalization to perform replaces on
3868 overlapping matches.
3870 return regex.sub(replacement, regex.sub(replacement, original))
3873 def re_compile_maybe_verbose(regex: str) -> Pattern[str]:
3874 """Compile a regular expression string in `regex`.
3876 If it contains newlines, use verbose mode.
3879 regex = "(?x)" + regex
3880 compiled: Pattern[str] = re.compile(regex)
3884 def enumerate_reversed(sequence: Sequence[T]) -> Iterator[Tuple[Index, T]]:
3885 """Like `reversed(enumerate(sequence))` if that were possible."""
3886 index = len(sequence) - 1
3887 for element in reversed(sequence):
3888 yield (index, element)
3892 def enumerate_with_length(
3893 line: Line, reversed: bool = False
3894 ) -> Iterator[Tuple[Index, Leaf, int]]:
3895 """Return an enumeration of leaves with their length.
3897 Stops prematurely on multiline strings and standalone comments.
3900 Callable[[Sequence[Leaf]], Iterator[Tuple[Index, Leaf]]],
3901 enumerate_reversed if reversed else enumerate,
3903 for index, leaf in op(line.leaves):
3904 length = len(leaf.prefix) + len(leaf.value)
3905 if "\n" in leaf.value:
3906 return # Multiline strings, we can't continue.
3908 for comment in line.comments_after(leaf):
3909 length += len(comment.value)
3911 yield index, leaf, length
3914 def is_line_short_enough(line: Line, *, line_length: int, line_str: str = "") -> bool:
3915 """Return True if `line` is no longer than `line_length`.
3917 Uses the provided `line_str` rendering, if any, otherwise computes a new one.
3920 line_str = str(line).strip("\n")
3922 len(line_str) <= line_length
3923 and "\n" not in line_str # multiline strings
3924 and not line.contains_standalone_comments()
3928 def can_be_split(line: Line) -> bool:
3929 """Return False if the line cannot be split *for sure*.
3931 This is not an exhaustive search but a cheap heuristic that we can use to
3932 avoid some unfortunate formattings (mostly around wrapping unsplittable code
3933 in unnecessary parentheses).
3935 leaves = line.leaves
3939 if leaves[0].type == token.STRING and leaves[1].type == token.DOT:
3943 for leaf in leaves[-2::-1]:
3944 if leaf.type in OPENING_BRACKETS:
3945 if next.type not in CLOSING_BRACKETS:
3949 elif leaf.type == token.DOT:
3951 elif leaf.type == token.NAME:
3952 if not (next.type == token.DOT or next.type in OPENING_BRACKETS):
3955 elif leaf.type not in CLOSING_BRACKETS:
3958 if dot_count > 1 and call_count > 1:
3964 def can_omit_invisible_parens(line: Line, line_length: int) -> bool:
3965 """Does `line` have a shape safe to reformat without optional parens around it?
3967 Returns True for only a subset of potentially nice looking formattings but
3968 the point is to not return false positives that end up producing lines that
3971 bt = line.bracket_tracker
3972 if not bt.delimiters:
3973 # Without delimiters the optional parentheses are useless.
3976 max_priority = bt.max_delimiter_priority()
3977 if bt.delimiter_count_with_priority(max_priority) > 1:
3978 # With more than one delimiter of a kind the optional parentheses read better.
3981 if max_priority == DOT_PRIORITY:
3982 # A single stranded method call doesn't require optional parentheses.
3985 assert len(line.leaves) >= 2, "Stranded delimiter"
3987 first = line.leaves[0]
3988 second = line.leaves[1]
3989 penultimate = line.leaves[-2]
3990 last = line.leaves[-1]
3992 # With a single delimiter, omit if the expression starts or ends with
3994 if first.type in OPENING_BRACKETS and second.type not in CLOSING_BRACKETS:
3996 length = 4 * line.depth
3997 for _index, leaf, leaf_length in enumerate_with_length(line):
3998 if leaf.type in CLOSING_BRACKETS and leaf.opening_bracket is first:
4001 length += leaf_length
4002 if length > line_length:
4005 if leaf.type in OPENING_BRACKETS:
4006 # There are brackets we can further split on.
4010 # checked the entire string and line length wasn't exceeded
4011 if len(line.leaves) == _index + 1:
4014 # Note: we are not returning False here because a line might have *both*
4015 # a leading opening bracket and a trailing closing bracket. If the
4016 # opening bracket doesn't match our rule, maybe the closing will.
4019 last.type == token.RPAR
4020 or last.type == token.RBRACE
4022 # don't use indexing for omitting optional parentheses;
4024 last.type == token.RSQB
4026 and last.parent.type != syms.trailer
4029 if penultimate.type in OPENING_BRACKETS:
4030 # Empty brackets don't help.
4033 if is_multiline_string(first):
4034 # Additional wrapping of a multiline string in this situation is
4038 length = 4 * line.depth
4039 seen_other_brackets = False
4040 for _index, leaf, leaf_length in enumerate_with_length(line):
4041 length += leaf_length
4042 if leaf is last.opening_bracket:
4043 if seen_other_brackets or length <= line_length:
4046 elif leaf.type in OPENING_BRACKETS:
4047 # There are brackets we can further split on.
4048 seen_other_brackets = True
4053 def get_cache_file(mode: FileMode) -> Path:
4054 return CACHE_DIR / f"cache.{mode.get_cache_key()}.pickle"
4057 def read_cache(mode: FileMode) -> Cache:
4058 """Read the cache if it exists and is well formed.
4060 If it is not well formed, the call to write_cache later should resolve the issue.
4062 cache_file = get_cache_file(mode)
4063 if not cache_file.exists():
4066 with cache_file.open("rb") as fobj:
4068 cache: Cache = pickle.load(fobj)
4069 except (pickle.UnpicklingError, ValueError):
4075 def get_cache_info(path: Path) -> CacheInfo:
4076 """Return the information used to check if a file is already formatted or not."""
4078 return stat.st_mtime, stat.st_size
4081 def filter_cached(cache: Cache, sources: Iterable[Path]) -> Tuple[Set[Path], Set[Path]]:
4082 """Split an iterable of paths in `sources` into two sets.
4084 The first contains paths of files that modified on disk or are not in the
4085 cache. The other contains paths to non-modified files.
4087 todo, done = set(), set()
4090 if cache.get(src) != get_cache_info(src):
4097 def write_cache(cache: Cache, sources: Iterable[Path], mode: FileMode) -> None:
4098 """Update the cache file."""
4099 cache_file = get_cache_file(mode)
4101 CACHE_DIR.mkdir(parents=True, exist_ok=True)
4102 new_cache = {**cache, **{src.resolve(): get_cache_info(src) for src in sources}}
4103 with tempfile.NamedTemporaryFile(dir=str(cache_file.parent), delete=False) as f:
4104 pickle.dump(new_cache, f, protocol=4)
4105 os.replace(f.name, cache_file)
4110 def patch_click() -> None:
4111 """Make Click not crash.
4113 On certain misconfigured environments, Python 3 selects the ASCII encoding as the
4114 default which restricts paths that it can access during the lifetime of the
4115 application. Click refuses to work in this scenario by raising a RuntimeError.
4117 In case of Black the likelihood that non-ASCII characters are going to be used in
4118 file paths is minimal since it's Python source code. Moreover, this crash was
4119 spurious on Python 3.7 thanks to PEP 538 and PEP 540.
4122 from click import core
4123 from click import _unicodefun # type: ignore
4124 except ModuleNotFoundError:
4127 for module in (core, _unicodefun):
4128 if hasattr(module, "_verify_python3_env"):
4129 module._verify_python3_env = lambda: None
4132 def patched_main() -> None:
4138 if __name__ == "__main__":