]> git.madduck.net Git - etc/vim.git/blob - black.py

madduck's git repository

Every one of the projects in this repository is available at the canonical URL git://git.madduck.net/madduck/pub/<projectpath> — see each project's metadata for the exact URL.

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.

SSH access, as well as push access can be individually arranged.

If you use my repositories frequently, consider adding the following snippet to ~/.gitconfig and using the third clone URL listed for each project:

[url "git://git.madduck.net/madduck/"]
  insteadOf = madduck:

3231503e4fb228c1bbbe40973ac73d47015c114e
[etc/vim.git] / black.py
1 import ast
2 import asyncio
3 from concurrent.futures import Executor, ProcessPoolExecutor
4 from contextlib import contextmanager
5 from datetime import datetime
6 from enum import Enum
7 from functools import lru_cache, partial, wraps
8 import io
9 import itertools
10 import logging
11 from multiprocessing import Manager, freeze_support
12 import os
13 from pathlib import Path
14 import pickle
15 import regex as re
16 import signal
17 import sys
18 import tempfile
19 import tokenize
20 import traceback
21 from typing import (
22     Any,
23     Callable,
24     Collection,
25     Dict,
26     Generator,
27     Generic,
28     Iterable,
29     Iterator,
30     List,
31     Optional,
32     Pattern,
33     Sequence,
34     Set,
35     Tuple,
36     TypeVar,
37     Union,
38     cast,
39 )
40
41 from appdirs import user_cache_dir
42 from attr import dataclass, evolve, Factory
43 import click
44 import toml
45 from typed_ast import ast3, ast27
46 from pathspec import PathSpec
47
48 # lib2to3 fork
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
54
55 from _black_version import version as __version__
56
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__))
61
62
63 # types
64 FileContent = str
65 Encoding = str
66 NewLine = str
67 Depth = int
68 NodeType = int
69 LeafID = int
70 Priority = int
71 Index = int
72 LN = Union[Leaf, Node]
73 SplitFunc = Callable[["Line", Collection["Feature"]], Iterator["Line"]]
74 Timestamp = float
75 FileSize = int
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)
80
81 pygram.initialize(CACHE_DIR)
82 syms = pygram.python_symbols
83
84
85 class NothingChanged(UserWarning):
86     """Raised when reformatted code is the same as source."""
87
88
89 class CannotSplit(Exception):
90     """A readable split that fits the allotted line length is impossible."""
91
92
93 class InvalidInput(ValueError):
94     """Raised when input source code fails all parse attempts."""
95
96
97 class WriteBack(Enum):
98     NO = 0
99     YES = 1
100     DIFF = 2
101     CHECK = 3
102
103     @classmethod
104     def from_configuration(cls, *, check: bool, diff: bool) -> "WriteBack":
105         if check and not diff:
106             return cls.CHECK
107
108         return cls.DIFF if diff else cls.YES
109
110
111 class Changed(Enum):
112     NO = 0
113     CACHED = 1
114     YES = 2
115
116
117 class TargetVersion(Enum):
118     PY27 = 2
119     PY33 = 3
120     PY34 = 4
121     PY35 = 5
122     PY36 = 6
123     PY37 = 7
124     PY38 = 8
125
126     def is_python2(self) -> bool:
127         return self is TargetVersion.PY27
128
129
130 PY36_VERSIONS = {TargetVersion.PY36, TargetVersion.PY37, TargetVersion.PY38}
131
132
133 class Feature(Enum):
134     # All string literals are unicode
135     UNICODE_LITERALS = 1
136     F_STRINGS = 2
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
143     ASYNC_KEYWORDS = 7
144     ASSIGNMENT_EXPRESSIONS = 8
145     POS_ONLY_ARGUMENTS = 9
146
147
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,
156     },
157     TargetVersion.PY36: {
158         Feature.UNICODE_LITERALS,
159         Feature.F_STRINGS,
160         Feature.NUMERIC_UNDERSCORES,
161         Feature.TRAILING_COMMA_IN_CALL,
162         Feature.TRAILING_COMMA_IN_DEF,
163         Feature.ASYNC_IDENTIFIERS,
164     },
165     TargetVersion.PY37: {
166         Feature.UNICODE_LITERALS,
167         Feature.F_STRINGS,
168         Feature.NUMERIC_UNDERSCORES,
169         Feature.TRAILING_COMMA_IN_CALL,
170         Feature.TRAILING_COMMA_IN_DEF,
171         Feature.ASYNC_KEYWORDS,
172     },
173     TargetVersion.PY38: {
174         Feature.UNICODE_LITERALS,
175         Feature.F_STRINGS,
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,
182     },
183 }
184
185
186 @dataclass
187 class FileMode:
188     target_versions: Set[TargetVersion] = Factory(set)
189     line_length: int = DEFAULT_LINE_LENGTH
190     string_normalization: bool = True
191     is_pyi: bool = False
192
193     def get_cache_key(self) -> str:
194         if self.target_versions:
195             version_str = ",".join(
196                 str(version.value)
197                 for version in sorted(self.target_versions, key=lambda v: v.value)
198             )
199         else:
200             version_str = "-"
201         parts = [
202             version_str,
203             str(self.line_length),
204             str(int(self.string_normalization)),
205             str(int(self.is_pyi)),
206         ]
207         return ".".join(parts)
208
209
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)
212
213
214 def read_pyproject_toml(
215     ctx: click.Context, param: click.Parameter, value: Union[str, int, bool, None]
216 ) -> Optional[str]:
217     """Inject Black configuration from "pyproject.toml" into defaults in `ctx`.
218
219     Returns the path to a successfully found and read configuration file, None
220     otherwise.
221     """
222     assert not isinstance(value, (int, bool)), "Invalid parameter type passed"
223     if not value:
224         root = find_project_root(ctx.params.get("src", ()))
225         path = root / "pyproject.toml"
226         if path.is_file():
227             value = str(path)
228         else:
229             return None
230
231     try:
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}"
237         )
238
239     if not config:
240         return None
241
242     if ctx.default_map is None:
243         ctx.default_map = {}
244     ctx.default_map.update(  # type: ignore  # bad types in .pyi
245         {k.replace("--", "").replace("-", "_"): v for k, v in config.items()}
246     )
247     return value
248
249
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.")
252 @click.option(
253     "-l",
254     "--line-length",
255     type=int,
256     default=DEFAULT_LINE_LENGTH,
257     help="How many characters per line to allow.",
258     show_default=True,
259 )
260 @click.option(
261     "-t",
262     "--target-version",
263     type=click.Choice([v.name.lower() for v in TargetVersion]),
264     callback=lambda c, p, v: [TargetVersion[val.upper()] for val in v],
265     multiple=True,
266     help=(
267         "Python versions that should be supported by Black's output. [default: "
268         "per-file auto-detection]"
269     ),
270 )
271 @click.option(
272     "--py36",
273     is_flag=True,
274     help=(
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]"
279     ),
280 )
281 @click.option(
282     "--pyi",
283     is_flag=True,
284     help=(
285         "Format all input files like typing stubs regardless of file extension "
286         "(useful when piping source on standard input)."
287     ),
288 )
289 @click.option(
290     "-S",
291     "--skip-string-normalization",
292     is_flag=True,
293     help="Don't normalize string quotes or prefixes.",
294 )
295 @click.option(
296     "--check",
297     is_flag=True,
298     help=(
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."
302     ),
303 )
304 @click.option(
305     "--diff",
306     is_flag=True,
307     help="Don't write the files back, just output a diff for each file on stdout.",
308 )
309 @click.option(
310     "--fast/--safe",
311     is_flag=True,
312     help="If --fast given, skip temporary sanity checks. [default: --safe]",
313 )
314 @click.option(
315     "--include",
316     type=str,
317     default=DEFAULT_INCLUDES,
318     help=(
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 "
323         "later."
324     ),
325     show_default=True,
326 )
327 @click.option(
328     "--exclude",
329     type=str,
330     default=DEFAULT_EXCLUDES,
331     help=(
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."
336     ),
337     show_default=True,
338 )
339 @click.option(
340     "-q",
341     "--quiet",
342     is_flag=True,
343     help=(
344         "Don't emit non-error messages to stderr. Errors are still emitted; "
345         "silence those with 2>/dev/null."
346     ),
347 )
348 @click.option(
349     "-v",
350     "--verbose",
351     is_flag=True,
352     help=(
353         "Also emit messages to stderr about files that were not changed or were "
354         "ignored due to --exclude=."
355     ),
356 )
357 @click.version_option(version=__version__)
358 @click.argument(
359     "src",
360     nargs=-1,
361     type=click.Path(
362         exists=True, file_okay=True, dir_okay=True, readable=True, allow_dash=True
363     ),
364     is_eager=True,
365 )
366 @click.option(
367     "--config",
368     type=click.Path(
369         exists=False, file_okay=True, dir_okay=False, readable=True, allow_dash=False
370     ),
371     is_eager=True,
372     callback=read_pyproject_toml,
373     help="Read configuration from PATH.",
374 )
375 @click.pass_context
376 def main(
377     ctx: click.Context,
378     code: Optional[str],
379     line_length: int,
380     target_version: List[TargetVersion],
381     check: bool,
382     diff: bool,
383     fast: bool,
384     pyi: bool,
385     py36: bool,
386     skip_string_normalization: bool,
387     quiet: bool,
388     verbose: bool,
389     include: str,
390     exclude: str,
391     src: Tuple[str],
392     config: Optional[str],
393 ) -> None:
394     """The uncompromising code formatter."""
395     write_back = WriteBack.from_configuration(check=check, diff=diff)
396     if target_version:
397         if py36:
398             err(f"Cannot use both --target-version and --py36")
399             ctx.exit(2)
400         else:
401             versions = set(target_version)
402     elif py36:
403         err(
404             "--py36 is deprecated and will be removed in a future version. "
405             "Use --target-version py36 instead."
406         )
407         versions = PY36_VERSIONS
408     else:
409         # We'll autodetect later.
410         versions = set()
411     mode = FileMode(
412         target_versions=versions,
413         line_length=line_length,
414         is_pyi=pyi,
415         string_normalization=not skip_string_normalization,
416     )
417     if config and verbose:
418         out(f"Using configuration from {config}.", bold=False, fg="blue")
419     if code is not None:
420         print(format_str(code, mode=mode))
421         ctx.exit(0)
422     try:
423         include_regex = re_compile_maybe_verbose(include)
424     except re.error:
425         err(f"Invalid regular expression for include given: {include!r}")
426         ctx.exit(2)
427     try:
428         exclude_regex = re_compile_maybe_verbose(exclude)
429     except re.error:
430         err(f"Invalid regular expression for exclude given: {exclude!r}")
431         ctx.exit(2)
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)
436     for s in src:
437         p = Path(s)
438         if p.is_dir():
439             sources.update(
440                 gen_python_files_in_dir(
441                     p, root, include_regex, exclude_regex, report, get_gitignore(root)
442                 )
443             )
444         elif p.is_file() or s == "-":
445             # if a file was explicitly given, we don't care about its extension
446             sources.add(p)
447         else:
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 😴")
452         ctx.exit(0)
453
454     if len(sources) == 1:
455         reformat_one(
456             src=sources.pop(),
457             fast=fast,
458             write_back=write_back,
459             mode=mode,
460             report=report,
461         )
462     else:
463         reformat_many(
464             sources=sources, fast=fast, write_back=write_back, mode=mode, report=report
465         )
466
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)
471
472
473 def path_empty(src: Tuple[str], quiet: bool, verbose: bool, ctx: click.Context) -> None:
474     """
475     Exit if there is no `src` provided for formatting
476     """
477     if not src:
478         if verbose or not quiet:
479             out("No Path provided. Nothing to do 😴")
480             ctx.exit(0)
481
482
483 def reformat_one(
484     src: Path, fast: bool, write_back: WriteBack, mode: FileMode, report: "Report"
485 ) -> None:
486     """Reformat a single file under `src` without spawning child processes.
487
488     `fast`, `write_back`, and `mode` options are passed to
489     :func:`format_file_in_place` or :func:`format_stdin_to_stdout`.
490     """
491     try:
492         changed = Changed.NO
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
496         else:
497             cache: Cache = {}
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
505             ):
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
509             ):
510                 write_cache(cache, [src], mode)
511         report.done(src, changed)
512     except Exception as exc:
513         report.failed(src, str(exc))
514
515
516 def reformat_many(
517     sources: Set[Path],
518     fast: bool,
519     write_back: WriteBack,
520     mode: FileMode,
521     report: "Report",
522 ) -> None:
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)
530     try:
531         loop.run_until_complete(
532             schedule_formatting(
533                 sources=sources,
534                 fast=fast,
535                 write_back=write_back,
536                 mode=mode,
537                 report=report,
538                 loop=loop,
539                 executor=executor,
540             )
541         )
542     finally:
543         shutdown(loop)
544         executor.shutdown()
545
546
547 async def schedule_formatting(
548     sources: Set[Path],
549     fast: bool,
550     write_back: WriteBack,
551     mode: FileMode,
552     report: "Report",
553     loop: asyncio.AbstractEventLoop,
554     executor: Executor,
555 ) -> None:
556     """Run formatting of `sources` in parallel using the provided `executor`.
557
558     (Use ProcessPoolExecutors for actual parallelism.)
559
560     `write_back`, `fast`, and `mode` options are passed to
561     :func:`format_file_in_place`.
562     """
563     cache: Cache = {}
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)
569     if not sources:
570         return
571
572     cancelled = []
573     sources_to_cache = []
574     lock = None
575     if write_back == WriteBack.DIFF:
576         # For diff output, we need locks to ensure we don't interleave output
577         # from different processes.
578         manager = Manager()
579         lock = manager.Lock()
580     tasks = {
581         asyncio.ensure_future(
582             loop.run_in_executor(
583                 executor, format_file_in_place, src, fast, mode, write_back, lock
584             )
585         ): src
586         for src in sorted(sources)
587     }
588     pending: Iterable[asyncio.Future] = tasks.keys()
589     try:
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.
594         pass
595     while pending:
596         done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
597         for task in done:
598             src = tasks.pop(task)
599             if task.cancelled():
600                 cancelled.append(task)
601             elif task.exception():
602                 report.failed(src, str(task.exception()))
603             else:
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
609                 ):
610                     sources_to_cache.append(src)
611                 report.done(src, changed)
612     if cancelled:
613         await asyncio.gather(*cancelled, loop=loop, return_exceptions=True)
614     if sources_to_cache:
615         write_cache(cache, sources_to_cache, mode)
616
617
618 def format_file_in_place(
619     src: Path,
620     fast: bool,
621     mode: FileMode,
622     write_back: WriteBack = WriteBack.NO,
623     lock: Any = None,  # multiprocessing.Manager().Lock() is some crazy proxy
624 ) -> bool:
625     """Format file under `src` path. Return True if changed.
626
627     If `write_back` is DIFF, write a diff to stdout. If it is YES, write reformatted
628     code to the file.
629     `mode` and `fast` options are passed to :func:`format_file_contents`.
630     """
631     if src.suffix == ".pyi":
632         mode = evolve(mode, is_pyi=True)
633
634     then = datetime.utcfromtimestamp(src.stat().st_mtime)
635     with open(src, "rb") as buf:
636         src_contents, encoding, newline = decode_bytes(buf.read())
637     try:
638         dst_contents = format_file_contents(src_contents, fast=fast, mode=mode)
639     except NothingChanged:
640         return False
641
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)
650
651         with lock or nullcontext():
652             f = io.TextIOWrapper(
653                 sys.stdout.buffer,
654                 encoding=encoding,
655                 newline=newline,
656                 write_through=True,
657             )
658             f.write(diff_contents)
659             f.detach()
660
661     return True
662
663
664 def format_stdin_to_stdout(
665     fast: bool, *, write_back: WriteBack = WriteBack.NO, mode: FileMode
666 ) -> bool:
667     """Format file on stdin. Return True if changed.
668
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`.
672     """
673     then = datetime.utcnow()
674     src, encoding, newline = decode_bytes(sys.stdin.buffer.read())
675     dst = src
676     try:
677         dst = format_file_contents(src, fast=fast, mode=mode)
678         return True
679
680     except NothingChanged:
681         return False
682
683     finally:
684         f = io.TextIOWrapper(
685             sys.stdout.buffer, encoding=encoding, newline=newline, write_through=True
686         )
687         if write_back == WriteBack.YES:
688             f.write(dst)
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))
694         f.detach()
695
696
697 def format_file_contents(
698     src_contents: str, *, fast: bool, mode: FileMode
699 ) -> FileContent:
700     """Reformat contents a file and return new contents.
701
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`.
705     """
706     if src_contents.strip() == "":
707         raise NothingChanged
708
709     dst_contents = format_str(src_contents, mode=mode)
710     if src_contents == dst_contents:
711         raise NothingChanged
712
713     if not fast:
714         assert_equivalent(src_contents, dst_contents)
715         assert_stable(src_contents, dst_contents, mode=mode)
716     return dst_contents
717
718
719 def format_str(src_contents: str, *, mode: FileMode) -> FileContent:
720     """Reformat a string and return new contents.
721
722     `mode` determines formatting options, such as how many characters per line are
723     allowed.
724     """
725     src_node = lib2to3_parse(src_contents.lstrip(), mode.target_versions)
726     dst_contents = []
727     future_imports = get_future_imports(src_node)
728     if mode.target_versions:
729         versions = mode.target_versions
730     else:
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),
736         is_pyi=mode.is_pyi,
737         normalize_strings=mode.string_normalization,
738     )
739     elt = EmptyLineTracker(is_pyi=mode.is_pyi)
740     empty_line = Line()
741     after = 0
742     split_line_features = {
743         feature
744         for feature in {Feature.TRAILING_COMMA_IN_CALL, Feature.TRAILING_COMMA_IN_DEF}
745         if supports_feature(versions, feature)
746     }
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
755         ):
756             dst_contents.append(str(line))
757     return "".join(dst_contents)
758
759
760 def decode_bytes(src: bytes) -> Tuple[FileContent, Encoding, NewLine]:
761     """Return a tuple of (decoded_contents, encoding, newline).
762
763     `newline` is either CRLF or LF but `decoded_contents` is decoded with
764     universal newlines (i.e. only contains LF).
765     """
766     srcbuf = io.BytesIO(src)
767     encoding, lines = tokenize.detect_encoding(srcbuf.readline)
768     if not lines:
769         return "", encoding, "\n"
770
771     newline = "\r\n" if b"\r\n" == lines[0][-2:] else "\n"
772     srcbuf.seek(0)
773     with io.TextIOWrapper(srcbuf, encoding) as tiow:
774         return tiow.read(), encoding, newline
775
776
777 def get_grammars(target_versions: Set[TargetVersion]) -> List[Grammar]:
778     if not target_versions:
779         # No target_version specified, so try all grammars.
780         return [
781             # Python 3.7+
782             pygram.python_grammar_no_print_statement_no_exec_statement_async_keywords,
783             # Python 3.0-3.6
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,
787             # Python 2.7
788             pygram.python_grammar,
789         ]
790
791     if all(version.is_python2() for version in target_versions):
792         # Python 2-only code, so try Python 2 grammars.
793         return [
794             # Python 2.7 with future print_function import
795             pygram.python_grammar_no_print_statement,
796             # Python 2.7
797             pygram.python_grammar,
798         ]
799
800     # Python 3-compatible code, so only try Python 3 grammar.
801     grammars = []
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):
804         # Python 3.7+
805         grammars.append(
806             pygram.python_grammar_no_print_statement_no_exec_statement_async_keywords
807         )
808     if not supports_feature(target_versions, Feature.ASYNC_KEYWORDS):
809         # Python 3.0-3.6
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
813     return grammars
814
815
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":
819         src_txt += "\n"
820
821     for grammar in get_grammars(set(target_versions)):
822         drv = driver.Driver(grammar, pytree.convert)
823         try:
824             result = drv.parse_string(src_txt, True)
825             break
826
827         except ParseError as pe:
828             lineno, column = pe.context[1]
829             lines = src_txt.splitlines()
830             try:
831                 faulty_line = lines[lineno - 1]
832             except IndexError:
833                 faulty_line = "<line number missing in source>"
834             exc = InvalidInput(f"Cannot parse: {lineno}:{column}: {faulty_line}")
835     else:
836         raise exc from None
837
838     if isinstance(result, Leaf):
839         result = Node(syms.file_input, [result])
840     return result
841
842
843 def lib2to3_unparse(node: Node) -> str:
844     """Given a lib2to3 node, return its string representation."""
845     code = str(node)
846     return code
847
848
849 T = TypeVar("T")
850
851
852 class Visitor(Generic[T]):
853     """Basic lib2to3 visitor that yields things of type `T` on `visit()`."""
854
855     def visit(self, node: LN) -> Iterator[T]:
856         """Main method to visit `node` and its children.
857
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()`
861         instead.
862
863         Then yields objects of type `T` from the selected visitor.
864         """
865         if node.type < 256:
866             name = token.tok_name[node.type]
867         else:
868             name = type_repr(node.type)
869         yield from getattr(self, f"visit_{name}", self.visit_default)(node)
870
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)
876
877
878 @dataclass
879 class DebugVisitor(Visitor[T]):
880     tree_depth: int = 0
881
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")
887             self.tree_depth += 1
888             for child in node.children:
889                 yield from self.visit(child)
890
891             self.tree_depth -= 1
892             out(f"{indent}/{_type}", fg="yellow", bold=False)
893         else:
894             _type = token.tok_name.get(node.type, str(node.type))
895             out(f"{indent}{_type}", fg="blue", nl=False)
896             if node.prefix:
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)
901
902     @classmethod
903     def show(cls, code: Union[str, Leaf, Node]) -> None:
904         """Pretty-print the lib2to3 AST of a given string of `code`.
905
906         Convenience method for debugging.
907         """
908         v: DebugVisitor[None] = DebugVisitor()
909         if isinstance(code, str):
910             code = lib2to3_parse(code)
911         list(v.visit(code))
912
913
914 WHITESPACE = {token.DEDENT, token.INDENT, token.NEWLINE}
915 STATEMENT = {
916     syms.if_stmt,
917     syms.while_stmt,
918     syms.for_stmt,
919     syms.try_stmt,
920     syms.except_clause,
921     syms.with_stmt,
922     syms.funcdef,
923     syms.classdef,
924 }
925 STANDALONE_COMMENT = 153
926 token.tok_name[STANDALONE_COMMENT] = "STANDALONE_COMMENT"
927 LOGIC_OPERATORS = {"and", "or"}
928 COMPARATORS = {
929     token.LESS,
930     token.GREATER,
931     token.EQEQUAL,
932     token.NOTEQUAL,
933     token.LESSEQUAL,
934     token.GREATEREQUAL,
935 }
936 MATH_OPERATORS = {
937     token.VBAR,
938     token.CIRCUMFLEX,
939     token.AMPER,
940     token.LEFTSHIFT,
941     token.RIGHTSHIFT,
942     token.PLUS,
943     token.MINUS,
944     token.STAR,
945     token.SLASH,
946     token.DOUBLESLASH,
947     token.PERCENT,
948     token.AT,
949     token.TILDE,
950     token.DOUBLESTAR,
951 }
952 STARS = {token.STAR, token.DOUBLESTAR}
953 VARARGS_SPECIALS = STARS | {token.SLASH}
954 VARARGS_PARENTS = {
955     syms.arglist,
956     syms.argument,  # double star in arglist
957     syms.trailer,  # single argument to call
958     syms.typedargslist,
959     syms.varargslist,  # lambdas
960 }
961 UNPACKING_PARENTS = {
962     syms.atom,  # single element of a list or set literal
963     syms.dictsetmaker,
964     syms.listmaker,
965     syms.testlist_gexp,
966     syms.testlist_star_expr,
967 }
968 TEST_DESCENDANTS = {
969     syms.test,
970     syms.lambdef,
971     syms.or_test,
972     syms.and_test,
973     syms.not_test,
974     syms.comparison,
975     syms.star_expr,
976     syms.expr,
977     syms.xor_expr,
978     syms.and_expr,
979     syms.shift_expr,
980     syms.arith_expr,
981     syms.trailer,
982     syms.term,
983     syms.power,
984 }
985 ASSIGNMENTS = {
986     "=",
987     "+=",
988     "-=",
989     "*=",
990     "@=",
991     "/=",
992     "%=",
993     "&=",
994     "|=",
995     "^=",
996     "<<=",
997     ">>=",
998     "**=",
999     "//=",
1000 }
1001 COMPREHENSION_PRIORITY = 20
1002 COMMA_PRIORITY = 18
1003 TERNARY_PRIORITY = 16
1004 LOGIC_PRIORITY = 14
1005 STRING_PRIORITY = 12
1006 COMPARATOR_PRIORITY = 10
1007 MATH_PRIORITIES = {
1008     token.VBAR: 9,
1009     token.CIRCUMFLEX: 8,
1010     token.AMPER: 7,
1011     token.LEFTSHIFT: 6,
1012     token.RIGHTSHIFT: 6,
1013     token.PLUS: 5,
1014     token.MINUS: 5,
1015     token.STAR: 4,
1016     token.SLASH: 4,
1017     token.DOUBLESLASH: 4,
1018     token.PERCENT: 4,
1019     token.AT: 4,
1020     token.TILDE: 3,
1021     token.DOUBLESTAR: 2,
1022 }
1023 DOT_PRIORITY = 1
1024
1025
1026 @dataclass
1027 class BracketTracker:
1028     """Keeps track of brackets on a line."""
1029
1030     depth: int = 0
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)
1036
1037     def mark(self, leaf: Leaf) -> None:
1038         """Mark `leaf` with bracket-related metadata. Keep track of delimiters.
1039
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.
1043
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.
1047
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
1050         `delimiters` field.
1051         """
1052         if leaf.type == token.COMMENT:
1053             return
1054
1055         self.maybe_decrement_after_for_loop_variable(leaf)
1056         self.maybe_decrement_after_lambda_arguments(leaf)
1057         if leaf.type in CLOSING_BRACKETS:
1058             self.depth -= 1
1059             opening_bracket = self.bracket_match.pop((self.depth, leaf.type))
1060             leaf.opening_bracket = opening_bracket
1061         leaf.bracket_depth = self.depth
1062         if self.depth == 0:
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
1066             else:
1067                 delim = is_split_after_delimiter(leaf, self.previous)
1068                 if delim:
1069                     self.delimiters[id(leaf)] = delim
1070         if leaf.type in OPENING_BRACKETS:
1071             self.bracket_match[self.depth, BRACKET[leaf.type]] = leaf
1072             self.depth += 1
1073         self.previous = leaf
1074         self.maybe_increment_lambda_arguments(leaf)
1075         self.maybe_increment_for_loop_variable(leaf)
1076
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)
1080
1081     def max_delimiter_priority(self, exclude: Iterable[LeafID] = ()) -> Priority:
1082         """Return the highest priority of a delimiter found on the line.
1083
1084         Values are consistent with what `is_split_*_delimiter()` return.
1085         Raises ValueError on no delimiters.
1086         """
1087         return max(v for k, v in self.delimiters.items() if k not in exclude)
1088
1089     def delimiter_count_with_priority(self, priority: Priority = 0) -> int:
1090         """Return the number of delimiters with the given `priority`.
1091
1092         If no `priority` is passed, defaults to max priority on the line.
1093         """
1094         if not self.delimiters:
1095             return 0
1096
1097         priority = priority or self.max_delimiter_priority()
1098         return sum(1 for p in self.delimiters.values() if p == priority)
1099
1100     def maybe_increment_for_loop_variable(self, leaf: Leaf) -> bool:
1101         """In a for loop, or comprehension, the variables are often unpacks.
1102
1103         To avoid splitting on the comma in this situation, increase the depth of
1104         tokens between `for` and `in`.
1105         """
1106         if leaf.type == token.NAME and leaf.value == "for":
1107             self.depth += 1
1108             self._for_loop_depths.append(self.depth)
1109             return True
1110
1111         return False
1112
1113     def maybe_decrement_after_for_loop_variable(self, leaf: Leaf) -> bool:
1114         """See `maybe_increment_for_loop_variable` above for explanation."""
1115         if (
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"
1120         ):
1121             self.depth -= 1
1122             self._for_loop_depths.pop()
1123             return True
1124
1125         return False
1126
1127     def maybe_increment_lambda_arguments(self, leaf: Leaf) -> bool:
1128         """In a lambda expression, there might be more than one argument.
1129
1130         To avoid splitting on the comma in this situation, increase the depth of
1131         tokens between `lambda` and `:`.
1132         """
1133         if leaf.type == token.NAME and leaf.value == "lambda":
1134             self.depth += 1
1135             self._lambda_argument_depths.append(self.depth)
1136             return True
1137
1138         return False
1139
1140     def maybe_decrement_after_lambda_arguments(self, leaf: Leaf) -> bool:
1141         """See `maybe_increment_lambda_arguments` above for explanation."""
1142         if (
1143             self._lambda_argument_depths
1144             and self._lambda_argument_depths[-1] == self.depth
1145             and leaf.type == token.COLON
1146         ):
1147             self.depth -= 1
1148             self._lambda_argument_depths.pop()
1149             return True
1150
1151         return False
1152
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))
1156
1157
1158 @dataclass
1159 class Line:
1160     """Holds leaves and comments. Can be printed with `str(line)`."""
1161
1162     depth: int = 0
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
1168
1169     def append(self, leaf: Leaf, preformatted: bool = False) -> None:
1170         """Add a new `leaf` to the end of the line.
1171
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.
1176
1177         Inline comments are put aside.
1178         """
1179         has_value = leaf.type in BRACKETS or bool(leaf.value.strip())
1180         if not has_value:
1181             return
1182
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)
1190             )
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)
1196
1197     def append_safe(self, leaf: Leaf, preformatted: bool = False) -> None:
1198         """Like :func:`append()` but disallow invalid standalone comment structure.
1199
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.
1202         """
1203         if self.bracket_tracker.depth == 0:
1204             if self.is_comment:
1205                 raise ValueError("cannot append to standalone comments")
1206
1207             if self.leaves and leaf.type == STANDALONE_COMMENT:
1208                 raise ValueError(
1209                     "cannot append standalone comments to a populated line"
1210                 )
1211
1212         self.append(leaf, preformatted=preformatted)
1213
1214     @property
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
1218
1219     @property
1220     def is_decorator(self) -> bool:
1221         """Is this line a decorator?"""
1222         return bool(self) and self.leaves[0].type == token.AT
1223
1224     @property
1225     def is_import(self) -> bool:
1226         """Is this an import line?"""
1227         return bool(self) and is_import(self.leaves[0])
1228
1229     @property
1230     def is_class(self) -> bool:
1231         """Is this line a class definition?"""
1232         return (
1233             bool(self)
1234             and self.leaves[0].type == token.NAME
1235             and self.leaves[0].value == "class"
1236         )
1237
1238     @property
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)
1243         ]
1244
1245     @property
1246     def is_collection_with_optional_trailing_comma(self) -> bool:
1247         """Is this line a collection literal with a trailing comma that's optional?
1248
1249         Note that the trailing comma in a 1-tuple is not optional.
1250         """
1251         if not self.leaves or len(self.leaves) < 4:
1252             return False
1253
1254         # Look for and address a trailing colon.
1255         if self.leaves[-1].type == token.COLON:
1256             closer = self.leaves[-2]
1257             close_index = -2
1258         else:
1259             closer = self.leaves[-1]
1260             close_index = -1
1261         if closer.type not in CLOSING_BRACKETS or self.inside_brackets:
1262             return False
1263
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
1267             # tuple.
1268             #
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):
1276                 if leaf is opener:
1277                     break
1278
1279             else:
1280                 # Couldn't find the matching opening paren, play it safe.
1281                 return False
1282
1283             commas = 0
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:
1287                     commas += 1
1288             if commas > 1:
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
1292
1293             elif commas == 1:
1294                 return False  # it's either a one-tuple or didn't have a trailing comma
1295
1296             if self.leaves[close_index - 1].type in CLOSING_BRACKETS:
1297                 close_index -= 1
1298                 closer = self.leaves[close_index]
1299                 if closer.type == token.RPAR:
1300                     # TODO: this is a gut feeling. Will we ever see this?
1301                     return False
1302
1303         if self.leaves[close_index - 1].type != token.COMMA:
1304             return False
1305
1306         return True
1307
1308     @property
1309     def is_def(self) -> bool:
1310         """Is this a function definition? (Also returns True for async defs.)"""
1311         try:
1312             first_leaf = self.leaves[0]
1313         except IndexError:
1314             return False
1315
1316         try:
1317             second_leaf: Optional[Leaf] = self.leaves[1]
1318         except IndexError:
1319             second_leaf = None
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"
1325         )
1326
1327     @property
1328     def is_class_paren_empty(self) -> bool:
1329         """Is this a class with no base classes but using parentheses?
1330
1331         Those are unnecessary and should be removed.
1332         """
1333         return (
1334             bool(self)
1335             and len(self.leaves) == 4
1336             and self.is_class
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 == ")"
1341         )
1342
1343     @property
1344     def is_triple_quoted_string(self) -> bool:
1345         """Is the line a triple quoted string?"""
1346         return (
1347             bool(self)
1348             and self.leaves[0].type == token.STRING
1349             and self.leaves[0].value.startswith(('"""', "'''"))
1350         )
1351
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:
1356                 return True
1357
1358         return False
1359
1360     def contains_uncollapsable_type_comments(self) -> bool:
1361         ignored_ids = set()
1362         try:
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
1367             ):
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))
1374         except IndexError:
1375             return False
1376
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
1381         # behind a comment.
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:
1387                         return True
1388
1389                 comment_seen = True
1390
1391         return False
1392
1393     def contains_unsplittable_type_ignore(self) -> bool:
1394         if not self.leaves:
1395             return False
1396
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.
1400         #
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.
1408
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)
1412
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
1416             # line.
1417             for node in self.leaves[-2:]:
1418                 for comment in self.comments.get(id(node), []):
1419                     if is_type_comment(comment, " ignore"):
1420                         return True
1421
1422         return False
1423
1424     def contains_multiline_strings(self) -> bool:
1425         for leaf in self.leaves:
1426             if is_multiline_string(leaf):
1427                 return True
1428
1429         return False
1430
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):
1434             return False
1435
1436         # We remove trailing commas only in the case of importing a
1437         # single name from a module.
1438         if not (
1439             self.leaves
1440             and self.is_import
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
1445             and (
1446                 # regular `from foo import bar,`
1447                 self.leaves[-4].value == "import"
1448                 # `from foo import (bar as baz,)
1449                 or (
1450                     len(self.leaves) > 6
1451                     and self.leaves[-6].value == "import"
1452                     and self.leaves[-3].value == "as"
1453                 )
1454                 # `from foo import bar as baz,`
1455                 or (
1456                     len(self.leaves) > 5
1457                     and self.leaves[-5].value == "import"
1458                     and self.leaves[-3].value == "as"
1459                 )
1460             )
1461             and closing.type == token.RPAR
1462         ):
1463             return False
1464
1465         self.remove_trailing_comma()
1466         return True
1467
1468     def append_comment(self, comment: Leaf) -> bool:
1469         """Add an inline or standalone comment to the line."""
1470         if (
1471             comment.type == STANDALONE_COMMENT
1472             and self.bracket_tracker.any_open_brackets()
1473         ):
1474             comment.prefix = ""
1475             return False
1476
1477         if comment.type != token.COMMENT:
1478             return False
1479
1480         if not self.leaves:
1481             comment.type = STANDALONE_COMMENT
1482             comment.prefix = ""
1483             return False
1484
1485         last_leaf = self.leaves[-1]
1486         if (
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)
1492         ):
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
1498                 comment.prefix = ""
1499                 return False
1500
1501             last_leaf = self.leaves[-2]
1502         self.comments.setdefault(id(last_leaf), []).append(comment)
1503         return True
1504
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), [])
1508
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
1515         )
1516
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:
1521             return False
1522
1523         subscript_start = open_lsqb.next_sibling
1524
1525         if isinstance(subscript_start, Node):
1526             if subscript_start.type == syms.listmaker:
1527                 return False
1528
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()
1533         )
1534
1535     def __str__(self) -> str:
1536         """Render the line."""
1537         if not self:
1538             return "\n"
1539
1540         indent = "    " * self.depth
1541         leaves = iter(self.leaves)
1542         first = next(leaves)
1543         res = f"{first.prefix}{indent}{first.value}"
1544         for leaf in leaves:
1545             res += str(leaf)
1546         for comment in itertools.chain.from_iterable(self.comments.values()):
1547             res += str(comment)
1548         return res + "\n"
1549
1550     def __bool__(self) -> bool:
1551         """Return True if the line has leaves or comments."""
1552         return bool(self.leaves or self.comments)
1553
1554
1555 @dataclass
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.
1559
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.
1563     """
1564
1565     is_pyi: bool = False
1566     previous_line: Optional[Line] = None
1567     previous_after: int = 0
1568     previous_defs: List[int] = Factory(list)
1569
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`.
1572
1573         This is for separating `def`, `async def` and `class` with extra empty
1574         lines (two on module-level).
1575         """
1576         before, after = self._maybe_empty_lines(current_line)
1577         before = (
1578             # Black should not insert empty lines at the beginning
1579             # of the file
1580             0
1581             if self.previous_line is None
1582             else before - self.previous_after
1583         )
1584         self.previous_after = after
1585         self.previous_line = current_line
1586         return before, after
1587
1588     def _maybe_empty_lines(self, current_line: Line) -> Tuple[int, int]:
1589         max_allowed = 1
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 = ""
1598         else:
1599             before = 0
1600         depth = current_line.depth
1601         while self.previous_defs and self.previous_defs[-1] >= depth:
1602             self.previous_defs.pop()
1603             if self.is_pyi:
1604                 before = 0 if depth else 1
1605             else:
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)
1609
1610         if (
1611             self.previous_line
1612             and self.previous_line.is_import
1613             and not current_line.is_import
1614             and depth == self.previous_line.depth
1615         ):
1616             return (before or 1), 0
1617
1618         if (
1619             self.previous_line
1620             and self.previous_line.is_class
1621             and current_line.is_triple_quoted_string
1622         ):
1623             return before, 1
1624
1625         return before, 0
1626
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.
1634             return 0, 0
1635
1636         if self.previous_line.is_decorator:
1637             return 0, 0
1638
1639         if self.previous_line.depth < current_line.depth and (
1640             self.previous_line.is_class or self.previous_line.is_def
1641         ):
1642             return 0, 0
1643
1644         if (
1645             self.previous_line.is_comment
1646             and self.previous_line.depth == current_line.depth
1647             and before == 0
1648         ):
1649             return 0, 0
1650
1651         if self.is_pyi:
1652             if self.previous_line.depth > current_line.depth:
1653                 newlines = 1
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
1657                     newlines = 0
1658                 else:
1659                     newlines = 1
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
1662                 newlines = 1
1663             else:
1664                 newlines = 0
1665         else:
1666             newlines = 2
1667         if current_line.depth and newlines:
1668             newlines -= 1
1669         return newlines, 0
1670
1671
1672 @dataclass
1673 class LineGenerator(Visitor[Line]):
1674     """Generates reformatted Line objects.  Empty lines are not emitted.
1675
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.
1678     """
1679
1680     is_pyi: bool = False
1681     normalize_strings: bool = True
1682     current_line: Line = Factory(Line)
1683     remove_u_prefix: bool = False
1684
1685     def line(self, indent: int = 0) -> Iterator[Line]:
1686         """Generate a line.
1687
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.
1690
1691         If any lines were generated, set up a new current_line.
1692         """
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.
1696
1697         complete_line = self.current_line
1698         self.current_line = Line(depth=complete_line.depth + indent)
1699         yield complete_line
1700
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()
1713
1714                 else:
1715                     # regular standalone comment
1716                     yield from self.line()
1717
1718                     self.current_line.append(comment)
1719                     yield from self.line()
1720
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)
1730
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)
1736
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()
1743
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)
1747
1748         # Finally, emit the dedent.
1749         yield from self.line(-1)
1750
1751     def visit_stmt(
1752         self, node: Node, keywords: Set[str], parens: Set[str]
1753     ) -> Iterator[Line]:
1754         """Visit a statement.
1755
1756         This implementation is shared for `if`, `while`, `for`, `try`, `except`,
1757         `def`, `with`, `class`, `assert` and assignments.
1758
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.
1761
1762         `parens` holds a set of string leaf values immediately after which
1763         invisible parens should be put.
1764         """
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()
1769
1770             yield from self.visit(child)
1771
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])
1776         else:
1777             yield from self.visit_default(node)
1778
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
1782         if is_suite_like:
1783             if self.is_pyi and is_stub_body(node):
1784                 yield from self.visit_default(node)
1785             else:
1786                 yield from self.line(+1)
1787                 yield from self.visit_default(node)
1788                 yield from self.line(-1)
1789
1790         else:
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)
1794
1795     def visit_async_stmt(self, node: Node) -> Iterator[Line]:
1796         """Visit `async def`, `async for`, `async with`."""
1797         yield from self.line()
1798
1799         children = iter(node.children)
1800         for child in children:
1801             yield from self.visit(child)
1802
1803             if child.type == token.ASYNC:
1804                 break
1805
1806         internal_stmt = next(children)
1807         for child in internal_stmt.children:
1808             yield from self.visit(child)
1809
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)
1815
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()
1819
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()
1824
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)
1829
1830     def visit_factor(self, node: Node) -> Iterator[Line]:
1831         """Force parentheses between a unary op and a binary power:
1832
1833         -2 ** 8 -> -(2 ** 8)
1834         """
1835         _operator, operand = node.children
1836         if (
1837             operand.type == syms.power
1838             and len(operand.children) == 3
1839             and operand.children[1].type == token.DOUBLESTAR
1840         ):
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)
1846
1847     def __attrs_post_init__(self) -> None:
1848         """You are in a twisty little maze of passages."""
1849         v = self.visit_stmt
1850         Ø: Set[str] = set()
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"}
1854         )
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=Ø
1859         )
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
1870
1871
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}
1878
1879
1880 def whitespace(leaf: Leaf, *, complex_subscript: bool) -> str:  # noqa: C901
1881     """Return whitespace prefix if needed for the given `leaf`.
1882
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.
1885     """
1886     NO = ""
1887     SPACE = " "
1888     DOUBLESPACE = "  "
1889     t = leaf.type
1890     p = leaf.parent
1891     v = leaf.value
1892     if t in ALWAYS_NO_SPACE:
1893         return NO
1894
1895     if t == token.COMMENT:
1896         return DOUBLESPACE
1897
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 {
1900         syms.subscript,
1901         syms.subscriptlist,
1902         syms.sliceop,
1903     }:
1904         return NO
1905
1906     prev = leaf.prev_sibling
1907     if not prev:
1908         prevp = preceding_leaf(p)
1909         if not prevp or prevp.type in OPENING_BRACKETS:
1910             return NO
1911
1912         if t == token.COLON:
1913             if prevp.type == token.COLON:
1914                 return NO
1915
1916             elif prevp.type != token.COMMA and not complex_subscript:
1917                 return NO
1918
1919             return SPACE
1920
1921         if prevp.type == token.EQUAL:
1922             if prevp.parent:
1923                 if prevp.parent.type in {
1924                     syms.arglist,
1925                     syms.argument,
1926                     syms.parameters,
1927                     syms.varargslist,
1928                 }:
1929                     return NO
1930
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
1934                     # that, too.
1935                     return prevp.prefix
1936
1937         elif prevp.type in VARARGS_SPECIALS:
1938             if is_vararg(prevp, within=VARARGS_PARENTS | UNPACKING_PARENTS):
1939                 return NO
1940
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
1944
1945         elif (
1946             prevp.parent
1947             and prevp.parent.type == syms.factor
1948             and prevp.type in MATH_OPERATORS
1949         ):
1950             return NO
1951
1952         elif (
1953             prevp.type == token.RIGHTSHIFT
1954             and prevp.parent
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
1959         ):
1960             # Python 2 print chevron
1961             return NO
1962
1963     elif prev.type in OPENING_BRACKETS:
1964         return NO
1965
1966     if p.type in {syms.parameters, syms.arglist}:
1967         # untyped function signatures or calls
1968         if not prev or prev.type != token.COMMA:
1969             return NO
1970
1971     elif p.type == syms.varargslist:
1972         # lambdas
1973         if prev and prev.type != token.COMMA:
1974             return NO
1975
1976     elif p.type == syms.typedargslist:
1977         # typed function signatures
1978         if not prev:
1979             return NO
1980
1981         if t == token.EQUAL:
1982             if prev.type != syms.tname:
1983                 return NO
1984
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.
1988             return prev.prefix
1989
1990         elif prev.type != token.COMMA:
1991             return NO
1992
1993     elif p.type == syms.tname:
1994         # type names
1995         if not prev:
1996             prevp = preceding_leaf(p)
1997             if not prevp or prevp.type != token.COMMA:
1998                 return NO
1999
2000     elif p.type == syms.trailer:
2001         # attributes and calls
2002         if t == token.LPAR or t == token.RPAR:
2003             return NO
2004
2005         if not prev:
2006             if t == token.DOT:
2007                 prevp = preceding_leaf(p)
2008                 if not prevp or prevp.type != token.NUMBER:
2009                     return NO
2010
2011             elif t == token.LSQB:
2012                 return NO
2013
2014         elif prev.type != token.COMMA:
2015             return NO
2016
2017     elif p.type == syms.argument:
2018         # single argument
2019         if t == token.EQUAL:
2020             return NO
2021
2022         if not prev:
2023             prevp = preceding_leaf(p)
2024             if not prevp or prevp.type == token.LPAR:
2025                 return NO
2026
2027         elif prev.type in {token.EQUAL} | VARARGS_SPECIALS:
2028             return NO
2029
2030     elif p.type == syms.decorator:
2031         # decorators
2032         return NO
2033
2034     elif p.type == syms.dotted_name:
2035         if prev:
2036             return NO
2037
2038         prevp = preceding_leaf(p)
2039         if not prevp or prevp.type == token.AT or prevp.type == token.DOT:
2040             return NO
2041
2042     elif p.type == syms.classdef:
2043         if t == token.LPAR:
2044             return NO
2045
2046         if prev and prev.type == token.LPAR:
2047             return NO
2048
2049     elif p.type in {syms.subscript, syms.sliceop}:
2050         # indexing
2051         if not prev:
2052             assert p.parent is not None, "subscripts are always parented"
2053             if p.parent.type == syms.subscriptlist:
2054                 return SPACE
2055
2056             return NO
2057
2058         elif not complex_subscript:
2059             return NO
2060
2061     elif p.type == syms.atom:
2062         if prev and t == token.DOT:
2063             # dots, but not the first one.
2064             return NO
2065
2066     elif p.type == syms.dictsetmaker:
2067         # dict unpacking
2068         if prev and prev.type == token.DOUBLESTAR:
2069             return NO
2070
2071     elif p.type in {syms.factor, syms.star_expr}:
2072         # unary ops
2073         if not prev:
2074             prevp = preceding_leaf(p)
2075             if not prevp or prevp.type in OPENING_BRACKETS:
2076                 return NO
2077
2078             prevp_parent = prevp.parent
2079             assert prevp_parent is not None
2080             if prevp.type == token.COLON and prevp_parent.type in {
2081                 syms.subscript,
2082                 syms.sliceop,
2083             }:
2084                 return NO
2085
2086             elif prevp.type == token.EQUAL and prevp_parent.type == syms.argument:
2087                 return NO
2088
2089         elif t in {token.NAME, token.NUMBER, token.STRING}:
2090             return NO
2091
2092     elif p.type == syms.import_from:
2093         if t == token.DOT:
2094             if prev and prev.type == token.DOT:
2095                 return NO
2096
2097         elif t == token.NAME:
2098             if v == "import":
2099                 return SPACE
2100
2101             if prev and prev.type == token.DOT:
2102                 return NO
2103
2104     elif p.type == syms.sliceop:
2105         return NO
2106
2107     return SPACE
2108
2109
2110 def preceding_leaf(node: Optional[LN]) -> Optional[Leaf]:
2111     """Return the first leaf that precedes `node`, if any."""
2112     while node:
2113         res = node.prev_sibling
2114         if res:
2115             if isinstance(res, Leaf):
2116                 return res
2117
2118             try:
2119                 return list(res.leaves())[-1]
2120
2121             except IndexError:
2122                 return None
2123
2124         node = node.parent
2125     return None
2126
2127
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:
2132         node = node.parent
2133     return node
2134
2135
2136 def container_of(leaf: Leaf) -> LN:
2137     """Return `leaf` or one of its ancestors that is the topmost container of it.
2138
2139     By "container" we mean a node where `leaf` is the very first child.
2140     """
2141     same_prefix = leaf.prefix
2142     container: LN = leaf
2143     while container:
2144         parent = container.parent
2145         if parent is None:
2146             break
2147
2148         if parent.children[0].prefix != same_prefix:
2149             break
2150
2151         if parent.type == syms.file_input:
2152             break
2153
2154         if parent.prev_sibling is not None and parent.prev_sibling.type in BRACKETS:
2155             break
2156
2157         container = parent
2158     return container
2159
2160
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.
2163
2164     The delimiter priorities returned here are from those delimiters that would
2165     cause a line break after themselves.
2166
2167     Higher numbers are higher priority.
2168     """
2169     if leaf.type == token.COMMA:
2170         return COMMA_PRIORITY
2171
2172     return 0
2173
2174
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.
2177
2178     The delimiter priorities returned here are from those delimiters that would
2179     cause a line break before themselves.
2180
2181     Higher numbers are higher priority.
2182     """
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.
2186         return 0
2187
2188     if (
2189         leaf.type == token.DOT
2190         and leaf.parent
2191         and leaf.parent.type not in {syms.import_from, syms.dotted_name}
2192         and (previous is None or previous.type in CLOSING_BRACKETS)
2193     ):
2194         return DOT_PRIORITY
2195
2196     if (
2197         leaf.type in MATH_OPERATORS
2198         and leaf.parent
2199         and leaf.parent.type not in {syms.factor, syms.star_expr}
2200     ):
2201         return MATH_PRIORITIES[leaf.type]
2202
2203     if leaf.type in COMPARATORS:
2204         return COMPARATOR_PRIORITY
2205
2206     if (
2207         leaf.type == token.STRING
2208         and previous is not None
2209         and previous.type == token.STRING
2210     ):
2211         return STRING_PRIORITY
2212
2213     if leaf.type not in {token.NAME, token.ASYNC}:
2214         return 0
2215
2216     if (
2217         leaf.value == "for"
2218         and leaf.parent
2219         and leaf.parent.type in {syms.comp_for, syms.old_comp_for}
2220         or leaf.type == token.ASYNC
2221     ):
2222         if (
2223             not isinstance(leaf.prev_sibling, Leaf)
2224             or leaf.prev_sibling.value != "async"
2225         ):
2226             return COMPREHENSION_PRIORITY
2227
2228     if (
2229         leaf.value == "if"
2230         and leaf.parent
2231         and leaf.parent.type in {syms.comp_if, syms.old_comp_if}
2232     ):
2233         return COMPREHENSION_PRIORITY
2234
2235     if leaf.value in {"if", "else"} and leaf.parent and leaf.parent.type == syms.test:
2236         return TERNARY_PRIORITY
2237
2238     if leaf.value == "is":
2239         return COMPARATOR_PRIORITY
2240
2241     if (
2242         leaf.value == "in"
2243         and leaf.parent
2244         and leaf.parent.type in {syms.comp_op, syms.comparison}
2245         and not (
2246             previous is not None
2247             and previous.type == token.NAME
2248             and previous.value == "not"
2249         )
2250     ):
2251         return COMPARATOR_PRIORITY
2252
2253     if (
2254         leaf.value == "not"
2255         and leaf.parent
2256         and leaf.parent.type == syms.comp_op
2257         and not (
2258             previous is not None
2259             and previous.type == token.NAME
2260             and previous.value == "is"
2261         )
2262     ):
2263         return COMPARATOR_PRIORITY
2264
2265     if leaf.value in LOGIC_OPERATORS and leaf.parent:
2266         return LOGIC_PRIORITY
2267
2268     return 0
2269
2270
2271 FMT_OFF = {"# fmt: off", "# fmt:off", "# yapf: disable"}
2272 FMT_ON = {"# fmt: on", "# fmt:on", "# yapf: enable"}
2273
2274
2275 def generate_comments(leaf: LN) -> Iterator[Leaf]:
2276     """Clean the prefix of the `leaf` and generate comments from it, if any.
2277
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.
2282
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.
2286
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.
2290
2291     Inline comments are emitted as regular token.COMMENT leaves.  Standalone
2292     are emitted with a fake STANDALONE_COMMENT token identifier.
2293     """
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)
2296
2297
2298 @dataclass
2299 class ProtoComment:
2300     """Describes a piece of syntax that is a comment.
2301
2302     It's not a :class:`blib2to3.pytree.Leaf` so that:
2303
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``.
2308     """
2309
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
2314
2315
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:
2321         return result
2322
2323     consumed = 0
2324     nlines = 0
2325     ignored_lines = 0
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()
2329         if not line:
2330             nlines += 1
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("\\"):
2336                 ignored_lines += 1
2337             continue
2338
2339         if index == ignored_lines and not is_endmarker:
2340             comment_type = token.COMMENT  # simple trailing comment
2341         else:
2342             comment_type = STANDALONE_COMMENT
2343         comment = make_comment(line)
2344         result.append(
2345             ProtoComment(
2346                 type=comment_type, value=comment, newlines=nlines, consumed=consumed
2347             )
2348         )
2349         nlines = 0
2350     return result
2351
2352
2353 def make_comment(content: str) -> str:
2354     """Return a consistently formatted comment from the given `content` string.
2355
2356     All comments (except for "##", "#!", "#:", '#'", "#%%") should have a single
2357     space between the hash sign and the content.
2358
2359     If `content` didn't start with a hash sign, one is provided.
2360     """
2361     content = content.rstrip()
2362     if not content:
2363         return "#"
2364
2365     if content[0] == "#":
2366         content = content[1:]
2367     if content and content[0] not in " !:#'%":
2368         content = " " + content
2369     return "#" + content
2370
2371
2372 def split_line(
2373     line: Line,
2374     line_length: int,
2375     inner: bool = False,
2376     features: Collection[Feature] = (),
2377 ) -> Iterator[Line]:
2378     """Split a `line` into potentially many lines.
2379
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.
2384
2385     `features` are syntactical features that may be used in the output.
2386     """
2387     if line.is_comment:
2388         yield line
2389         return
2390
2391     line_str = str(line).strip("\n")
2392
2393     if (
2394         not line.contains_uncollapsable_type_comments()
2395         and not line.should_explode
2396         and not line.is_collection_with_optional_trailing_comma
2397         and (
2398             is_line_short_enough(line, line_length=line_length, line_str=line_str)
2399             or line.contains_unsplittable_type_ignore()
2400         )
2401     ):
2402         yield line
2403         return
2404
2405     split_funcs: List[SplitFunc]
2406     if line.is_def:
2407         split_funcs = [left_hand_split]
2408     else:
2409
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):
2414                     yield from lines
2415                     return
2416
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)
2423
2424         if line.inside_brackets:
2425             split_funcs = [delimiter_split, standalone_comment_split, rhs]
2426         else:
2427             split_funcs = [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
2431         # split altogether.
2432         result: List[Line] = []
2433         try:
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")
2437
2438                 result.extend(
2439                     split_line(
2440                         l, line_length=line_length, inner=True, features=features
2441                     )
2442                 )
2443         except CannotSplit:
2444             continue
2445
2446         else:
2447             yield from result
2448             break
2449
2450     else:
2451         yield line
2452
2453
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.
2456
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.
2460     """
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:
2467         if (
2468             current_leaves is body_leaves
2469             and leaf.type in CLOSING_BRACKETS
2470             and leaf.opening_bracket is matching_bracket
2471         ):
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")
2480
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):
2486         if result:
2487             yield result
2488
2489
2490 def right_hand_split(
2491     line: Line,
2492     line_length: int,
2493     features: Collection[Feature] = (),
2494     omit: Collection[LeafID] = (),
2495 ) -> Iterator[Line]:
2496     """Split line into many lines, starting with the last matching bracket pair.
2497
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
2500     this split.
2501
2502     Note: running this function modifies `bracket_depth` on the leaves of `line`.
2503     """
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")
2525
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)
2533     if (
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)
2549     ):
2550         omit = {id(closing_bracket), *omit}
2551         try:
2552             yield from right_hand_split(line, line_length, features=features, omit=omit)
2553             return
2554
2555         except CannotSplit:
2556             if not (
2557                 can_be_split(body)
2558                 or is_line_short_enough(body, line_length=line_length)
2559             ):
2560                 raise CannotSplit(
2561                     "Splitting failed, body is still too long and can't be split."
2562                 )
2563
2564             elif head.contains_multiline_strings() or tail.contains_multiline_strings():
2565                 raise CannotSplit(
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 "
2569                     "line."
2570                 )
2571
2572     ensure_visible(opening_bracket)
2573     ensure_visible(closing_bracket)
2574     for result in (head, body, tail):
2575         if result:
2576             yield result
2577
2578
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.
2581
2582     Do nothing otherwise.
2583
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.
2588
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.
2592     """
2593     tail_len = len(str(tail).strip())
2594     if not body:
2595         if tail_len == 0:
2596             raise CannotSplit("Splitting brackets produced the same line")
2597
2598         elif tail_len < 3:
2599             raise CannotSplit(
2600                 f"Splitting brackets on an empty body to save "
2601                 f"{tail_len} characters is not worth it"
2602             )
2603
2604
2605 def bracket_split_build_line(
2606     leaves: List[Leaf], original: Line, opening_bracket: Leaf, *, is_body: bool = False
2607 ) -> Line:
2608     """Return a new line with given `leaves` and respective comments from `original`.
2609
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.
2612     """
2613     result = Line(depth=original.depth)
2614     if is_body:
2615         result.inside_brackets = True
2616         result.depth += 1
2617         if leaves:
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.
2622             no_commas = (
2623                 original.is_def
2624                 and opening_bracket.value == "("
2625                 and not any(l.type == token.COMMA for l in leaves)
2626             )
2627
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:
2631                         continue
2632
2633                     if leaves[i].type != token.COMMA:
2634                         leaves.insert(i + 1, Leaf(token.COMMA, ","))
2635                     break
2636
2637     # Populate the line
2638     for leaf in leaves:
2639         result.append(leaf, preformatted=True)
2640         for comment_after in original.comments_after(leaf):
2641             result.append(comment_after, preformatted=True)
2642     if is_body:
2643         result.should_explode = should_explode(result, opening_bracket)
2644     return result
2645
2646
2647 def dont_increase_indentation(split_func: SplitFunc) -> SplitFunc:
2648     """Normalize prefix of the first leaf in every line returned by `split_func`.
2649
2650     This is a decorator over relevant split functions.
2651     """
2652
2653     @wraps(split_func)
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)
2657             yield l
2658
2659     return split_wrapper
2660
2661
2662 @dont_increase_indentation
2663 def delimiter_split(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2664     """Split according to delimiters of the highest priority.
2665
2666     If the appropriate Features are given, the split will add trailing commas
2667     also in function signatures and calls that contain `*` and `**`.
2668     """
2669     try:
2670         last_leaf = line.leaves[-1]
2671     except IndexError:
2672         raise CannotSplit("Line empty")
2673
2674     bt = line.bracket_tracker
2675     try:
2676         delimiter_priority = bt.max_delimiter_priority(exclude={id(last_leaf)})
2677     except ValueError:
2678         raise CannotSplit("No delimiters found")
2679
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")
2683
2684     current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2685     lowest_depth = sys.maxsize
2686     trailing_comma_safe = True
2687
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
2691         try:
2692             current_line.append_safe(leaf, preformatted=True)
2693         except ValueError:
2694             yield current_line
2695
2696             current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2697             current_line.append(leaf)
2698
2699     for leaf in line.leaves:
2700         yield from append_to_line(leaf)
2701
2702         for comment_after in line.comments_after(leaf):
2703             yield from append_to_line(comment_after)
2704
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
2710                 )
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
2714                 )
2715
2716         leaf_priority = bt.delimiters.get(id(leaf))
2717         if leaf_priority == delimiter_priority:
2718             yield current_line
2719
2720             current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2721     if current_line:
2722         if (
2723             trailing_comma_safe
2724             and delimiter_priority == COMMA_PRIORITY
2725             and current_line.leaves[-1].type != token.COMMA
2726             and current_line.leaves[-1].type != STANDALONE_COMMENT
2727         ):
2728             current_line.append(Leaf(token.COMMA, ","))
2729         yield current_line
2730
2731
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")
2739
2740     current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2741
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
2745         try:
2746             current_line.append_safe(leaf, preformatted=True)
2747         except ValueError:
2748             yield current_line
2749
2750             current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2751             current_line.append(leaf)
2752
2753     for leaf in line.leaves:
2754         yield from append_to_line(leaf)
2755
2756         for comment_after in line.comments_after(leaf):
2757             yield from append_to_line(comment_after)
2758
2759     if current_line:
2760         yield current_line
2761
2762
2763 def is_import(leaf: Leaf) -> bool:
2764     """Return True if the given leaf starts an import statement."""
2765     p = leaf.parent
2766     t = leaf.type
2767     v = leaf.value
2768     return bool(
2769         t == token.NAME
2770         and (
2771             (v == "import" and p and p.type == syms.import_name)
2772             or (v == "from" and p and p.type == syms.import_from)
2773         )
2774     )
2775
2776
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."""
2780     t = leaf.type
2781     v = leaf.value
2782     return t in {token.COMMENT, STANDALONE_COMMENT} and v.startswith("# type:" + suffix)
2783
2784
2785 def normalize_prefix(leaf: Leaf, *, inside_brackets: bool) -> None:
2786     """Leave existing extra newlines if not `inside_brackets`. Remove everything
2787     else.
2788
2789     Note: don't use backslashes for formatting or you'll lose your voting rights.
2790     """
2791     if not inside_brackets:
2792         spl = leaf.prefix.split("#")
2793         if "\\" not in spl[0]:
2794             nl_count = spl[-1].count("\n")
2795             if len(spl) > 1:
2796                 nl_count -= 1
2797             leaf.prefix = "\n" * nl_count
2798             return
2799
2800     leaf.prefix = ""
2801
2802
2803 def normalize_string_prefix(leaf: Leaf, remove_u_prefix: bool = False) -> None:
2804     """Make all string prefixes lowercase.
2805
2806     If remove_u_prefix is given, also removes any u prefix from the string.
2807
2808     Note: Mutates its argument.
2809     """
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()
2814     if remove_u_prefix:
2815         new_prefix = new_prefix.replace("u", "")
2816     leaf.value = f"{new_prefix}{match.group(2)}"
2817
2818
2819 def normalize_string_quotes(leaf: Leaf) -> None:
2820     """Prefer double quotes but only if it doesn't cause more escaping.
2821
2822     Adds or removes backslashes as appropriate. Doesn't parse and fix
2823     strings nested in f-strings (yet).
2824
2825     Note: Mutates its argument.
2826     """
2827     value = leaf.value.lstrip("furbFURB")
2828     if value[:3] == '"""':
2829         return
2830
2831     elif value[:3] == "'''":
2832         orig_quote = "'''"
2833         new_quote = '"""'
2834     elif value[0] == '"':
2835         orig_quote = '"'
2836         new_quote = "'"
2837     else:
2838         orig_quote = "'"
2839         new_quote = '"'
2840     first_quote_pos = leaf.value.find(orig_quote)
2841     if first_quote_pos == -1:
2842         return  # There's an internal error
2843
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
2853             return
2854
2855         # Do not introduce or remove backslashes in raw strings
2856         new_body = body
2857     else:
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
2862             body = new_body
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(
2868             r"""
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-}
2872             """,
2873             new_body,
2874             re.VERBOSE,
2875         )
2876         for m in matches:
2877             if "\\" in str(m):
2878                 # Do not introduce backslashes in interpolated expressions
2879                 return
2880
2881     if new_quote == '"""' and new_body[-1:] == '"':
2882         # edge case:
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
2888
2889     if new_escape_count == orig_escape_count and orig_quote == '"':
2890         return  # Prefer double quotes
2891
2892     leaf.value = f"{prefix}{new_quote}{new_body}{new_quote}"
2893
2894
2895 def normalize_numeric_literal(leaf: Leaf) -> None:
2896     """Normalizes numeric (float, int, and complex) literals.
2897
2898     All letters used in the representation are normalized to lowercase (except
2899     in Python 2 long literals).
2900     """
2901     text = leaf.value.lower()
2902     if text.startswith(("0o", "0b")):
2903         # Leave octal and binary literals alone.
2904         pass
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()}"
2909     elif "e" in text:
2910         before, after = text.split("e")
2911         sign = ""
2912         if after.startswith("-"):
2913             after = after[1:]
2914             sign = "-"
2915         elif after.startswith("+"):
2916             after = after[1:]
2917         before = format_float_or_int_string(before)
2918         text = f"{before}e{sign}{after}"
2919     elif text.endswith(("j", "l")):
2920         number = text[:-1]
2921         suffix = text[-1]
2922         # Capitalize in "2L" because "l" looks too similar to "1".
2923         if suffix == "l":
2924             suffix = "L"
2925         text = f"{format_float_or_int_string(number)}{suffix}"
2926     else:
2927         text = format_float_or_int_string(text)
2928     leaf.value = text
2929
2930
2931 def format_float_or_int_string(text: str) -> str:
2932     """Formats a float string like "1.0"."""
2933     if "." not in text:
2934         return text
2935
2936     before, after = text.split(".")
2937     return f"{before or 0}.{after or 0}"
2938
2939
2940 def normalize_invisible_parens(node: Node, parens_after: Set[str]) -> None:
2941     """Make existing optional parentheses invisible or create new ones.
2942
2943     `parens_after` is a set of string leaf values immediately after which parens
2944     should be put.
2945
2946     Standardizes on visible parentheses for single-element tuples, and keeps
2947     existing visible parentheses for other tuples and generator expressions.
2948     """
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.
2952             return
2953     check_lpar = False
2954     for index, child in enumerate(list(node.children)):
2955         # Add parentheses around long tuple unpacking in assignments.
2956         if (
2957             index == 0
2958             and isinstance(child, Node)
2959             and child.type == syms.testlist_star_expr
2960         ):
2961             check_lpar = True
2962
2963         if check_lpar:
2964             if is_walrus_assignment(child):
2965                 continue
2966
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
2974                 # the statement
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, ""))
2983                 break
2984
2985             elif not (isinstance(child, Leaf) and is_multiline_string(child)):
2986                 wrap_in_parentheses(node, child, visible=False)
2987
2988         check_lpar = isinstance(child, Leaf) and child.value in parens_after
2989
2990
2991 def normalize_fmt_off(node: Node) -> None:
2992     """Convert content between `# fmt: off`/`# fmt: on` into standalone comments."""
2993     try_again = True
2994     while try_again:
2995         try_again = convert_one_fmt_off_pair(node)
2996
2997
2998 def convert_one_fmt_off_pair(node: Node) -> bool:
2999     """Convert content of a single `# fmt: off`/`# fmt: on` into a standalone comment.
3000
3001     Returns True if a pair was converted.
3002     """
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
3009                 # disguise.
3010                 if comment.type != STANDALONE_COMMENT:
3011                     prev = preceding_leaf(leaf)
3012                     if prev and prev.type not in WHITESPACE:
3013                         continue
3014
3015                 ignored_nodes = list(generate_ignored_nodes(leaf))
3016                 if not ignored_nodes:
3017                     continue
3018
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 :]
3023                 hidden_value = (
3024                     comment.value + "\n" + "".join(str(n) for n in ignored_nodes)
3025                 )
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]
3030                 first_idx = None
3031                 for ignored in ignored_nodes:
3032                     index = ignored.remove()
3033                     if first_idx is None:
3034                         first_idx = index
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(
3038                     first_idx,
3039                     Leaf(
3040                         STANDALONE_COMMENT,
3041                         hidden_value,
3042                         prefix=prefix[:previous_consumed] + "\n" * comment.newlines,
3043                     ),
3044                 )
3045                 return True
3046
3047             previous_consumed = comment.consumed
3048
3049     return False
3050
3051
3052 def generate_ignored_nodes(leaf: Leaf) -> Iterator[LN]:
3053     """Starting from the container of `leaf`, generate all leaves until `# fmt: on`.
3054
3055     Stops at the end of the block.
3056     """
3057     container: Optional[LN] = container_of(leaf)
3058     while container is not None and container.type != token.ENDMARKER:
3059         for comment in list_comments(container.prefix, is_endmarker=False):
3060             if comment.value in FMT_ON:
3061                 return
3062
3063         yield container
3064
3065         container = container.next_sibling
3066
3067
3068 def maybe_make_parens_invisible_in_atom(node: LN, parent: LN) -> bool:
3069     """If it's safe, make the parens in the atom `node` invisible, recursively.
3070     Additionally, remove repeated, adjacent invisible parens from the atom `node`
3071     as they are redundant.
3072
3073     Returns whether the node should itself be wrapped in invisible parentheses.
3074
3075     """
3076     if (
3077         node.type != syms.atom
3078         or is_empty_tuple(node)
3079         or is_one_tuple(node)
3080         or (is_yield(node) and parent.type != syms.expr_stmt)
3081         or max_delimiter_priority_in_atom(node) >= COMMA_PRIORITY
3082     ):
3083         return False
3084
3085     first = node.children[0]
3086     last = node.children[-1]
3087     if first.type == token.LPAR and last.type == token.RPAR:
3088         middle = node.children[1]
3089         # make parentheses invisible
3090         first.value = ""  # type: ignore
3091         last.value = ""  # type: ignore
3092         maybe_make_parens_invisible_in_atom(middle, parent=parent)
3093
3094         if is_atom_with_invisible_parens(middle):
3095             # Strip the invisible parens from `middle` by replacing
3096             # it with the child in-between the invisible parens
3097             middle.replace(middle.children[1])
3098
3099         return False
3100
3101     return True
3102
3103
3104 def is_atom_with_invisible_parens(node: LN) -> bool:
3105     """Given a `LN`, determines whether it's an atom `node` with invisible
3106     parens. Useful in dedupe-ing and normalizing parens.
3107     """
3108     if isinstance(node, Leaf) or node.type != syms.atom:
3109         return False
3110
3111     first, last = node.children[0], node.children[-1]
3112     return (
3113         isinstance(first, Leaf)
3114         and first.type == token.LPAR
3115         and first.value == ""
3116         and isinstance(last, Leaf)
3117         and last.type == token.RPAR
3118         and last.value == ""
3119     )
3120
3121
3122 def is_empty_tuple(node: LN) -> bool:
3123     """Return True if `node` holds an empty tuple."""
3124     return (
3125         node.type == syms.atom
3126         and len(node.children) == 2
3127         and node.children[0].type == token.LPAR
3128         and node.children[1].type == token.RPAR
3129     )
3130
3131
3132 def unwrap_singleton_parenthesis(node: LN) -> Optional[LN]:
3133     """Returns `wrapped` if `node` is of the shape ( wrapped ).
3134
3135     Parenthesis can be optional. Returns None otherwise"""
3136     if len(node.children) != 3:
3137         return None
3138
3139     lpar, wrapped, rpar = node.children
3140     if not (lpar.type == token.LPAR and rpar.type == token.RPAR):
3141         return None
3142
3143     return wrapped
3144
3145
3146 def wrap_in_parentheses(parent: Node, child: LN, *, visible: bool = True) -> None:
3147     """Wrap `child` in parentheses.
3148
3149     This replaces `child` with an atom holding the parentheses and the old
3150     child.  That requires moving the prefix.
3151
3152     If `visible` is False, the leaves will be valueless (and thus invisible).
3153     """
3154     lpar = Leaf(token.LPAR, "(" if visible else "")
3155     rpar = Leaf(token.RPAR, ")" if visible else "")
3156     prefix = child.prefix
3157     child.prefix = ""
3158     index = child.remove() or 0
3159     new_child = Node(syms.atom, [lpar, child, rpar])
3160     new_child.prefix = prefix
3161     parent.insert_child(index, new_child)
3162
3163
3164 def is_one_tuple(node: LN) -> bool:
3165     """Return True if `node` holds a tuple with one element, with or without parens."""
3166     if node.type == syms.atom:
3167         gexp = unwrap_singleton_parenthesis(node)
3168         if gexp is None or gexp.type != syms.testlist_gexp:
3169             return False
3170
3171         return len(gexp.children) == 2 and gexp.children[1].type == token.COMMA
3172
3173     return (
3174         node.type in IMPLICIT_TUPLE
3175         and len(node.children) == 2
3176         and node.children[1].type == token.COMMA
3177     )
3178
3179
3180 def is_walrus_assignment(node: LN) -> bool:
3181     """Return True iff `node` is of the shape ( test := test )"""
3182     inner = unwrap_singleton_parenthesis(node)
3183     return inner is not None and inner.type == syms.namedexpr_test
3184
3185
3186 def is_yield(node: LN) -> bool:
3187     """Return True if `node` holds a `yield` or `yield from` expression."""
3188     if node.type == syms.yield_expr:
3189         return True
3190
3191     if node.type == token.NAME and node.value == "yield":  # type: ignore
3192         return True
3193
3194     if node.type != syms.atom:
3195         return False
3196
3197     if len(node.children) != 3:
3198         return False
3199
3200     lpar, expr, rpar = node.children
3201     if lpar.type == token.LPAR and rpar.type == token.RPAR:
3202         return is_yield(expr)
3203
3204     return False
3205
3206
3207 def is_vararg(leaf: Leaf, within: Set[NodeType]) -> bool:
3208     """Return True if `leaf` is a star or double star in a vararg or kwarg.
3209
3210     If `within` includes VARARGS_PARENTS, this applies to function signatures.
3211     If `within` includes UNPACKING_PARENTS, it applies to right hand-side
3212     extended iterable unpacking (PEP 3132) and additional unpacking
3213     generalizations (PEP 448).
3214     """
3215     if leaf.type not in VARARGS_SPECIALS or not leaf.parent:
3216         return False
3217
3218     p = leaf.parent
3219     if p.type == syms.star_expr:
3220         # Star expressions are also used as assignment targets in extended
3221         # iterable unpacking (PEP 3132).  See what its parent is instead.
3222         if not p.parent:
3223             return False
3224
3225         p = p.parent
3226
3227     return p.type in within
3228
3229
3230 def is_multiline_string(leaf: Leaf) -> bool:
3231     """Return True if `leaf` is a multiline string that actually spans many lines."""
3232     value = leaf.value.lstrip("furbFURB")
3233     return value[:3] in {'"""', "'''"} and "\n" in value
3234
3235
3236 def is_stub_suite(node: Node) -> bool:
3237     """Return True if `node` is a suite with a stub body."""
3238     if (
3239         len(node.children) != 4
3240         or node.children[0].type != token.NEWLINE
3241         or node.children[1].type != token.INDENT
3242         or node.children[3].type != token.DEDENT
3243     ):
3244         return False
3245
3246     return is_stub_body(node.children[2])
3247
3248
3249 def is_stub_body(node: LN) -> bool:
3250     """Return True if `node` is a simple statement containing an ellipsis."""
3251     if not isinstance(node, Node) or node.type != syms.simple_stmt:
3252         return False
3253
3254     if len(node.children) != 2:
3255         return False
3256
3257     child = node.children[0]
3258     return (
3259         child.type == syms.atom
3260         and len(child.children) == 3
3261         and all(leaf == Leaf(token.DOT, ".") for leaf in child.children)
3262     )
3263
3264
3265 def max_delimiter_priority_in_atom(node: LN) -> Priority:
3266     """Return maximum delimiter priority inside `node`.
3267
3268     This is specific to atoms with contents contained in a pair of parentheses.
3269     If `node` isn't an atom or there are no enclosing parentheses, returns 0.
3270     """
3271     if node.type != syms.atom:
3272         return 0
3273
3274     first = node.children[0]
3275     last = node.children[-1]
3276     if not (first.type == token.LPAR and last.type == token.RPAR):
3277         return 0
3278
3279     bt = BracketTracker()
3280     for c in node.children[1:-1]:
3281         if isinstance(c, Leaf):
3282             bt.mark(c)
3283         else:
3284             for leaf in c.leaves():
3285                 bt.mark(leaf)
3286     try:
3287         return bt.max_delimiter_priority()
3288
3289     except ValueError:
3290         return 0
3291
3292
3293 def ensure_visible(leaf: Leaf) -> None:
3294     """Make sure parentheses are visible.
3295
3296     They could be invisible as part of some statements (see
3297     :func:`normalize_invisible_parens` and :func:`visit_import_from`).
3298     """
3299     if leaf.type == token.LPAR:
3300         leaf.value = "("
3301     elif leaf.type == token.RPAR:
3302         leaf.value = ")"
3303
3304
3305 def should_explode(line: Line, opening_bracket: Leaf) -> bool:
3306     """Should `line` immediately be split with `delimiter_split()` after RHS?"""
3307
3308     if not (
3309         opening_bracket.parent
3310         and opening_bracket.parent.type in {syms.atom, syms.import_from}
3311         and opening_bracket.value in "[{("
3312     ):
3313         return False
3314
3315     try:
3316         last_leaf = line.leaves[-1]
3317         exclude = {id(last_leaf)} if last_leaf.type == token.COMMA else set()
3318         max_priority = line.bracket_tracker.max_delimiter_priority(exclude=exclude)
3319     except (IndexError, ValueError):
3320         return False
3321
3322     return max_priority == COMMA_PRIORITY
3323
3324
3325 def get_features_used(node: Node) -> Set[Feature]:
3326     """Return a set of (relatively) new Python features used in this file.
3327
3328     Currently looking for:
3329     - f-strings;
3330     - underscores in numeric literals;
3331     - trailing commas after * or ** in function signatures and calls;
3332     - positional only arguments in function signatures and lambdas;
3333     """
3334     features: Set[Feature] = set()
3335     for n in node.pre_order():
3336         if n.type == token.STRING:
3337             value_head = n.value[:2]  # type: ignore
3338             if value_head in {'f"', 'F"', "f'", "F'", "rf", "fr", "RF", "FR"}:
3339                 features.add(Feature.F_STRINGS)
3340
3341         elif n.type == token.NUMBER:
3342             if "_" in n.value:  # type: ignore
3343                 features.add(Feature.NUMERIC_UNDERSCORES)
3344
3345         elif n.type == token.SLASH:
3346             if n.parent and n.parent.type in {syms.typedargslist, syms.arglist}:
3347                 features.add(Feature.POS_ONLY_ARGUMENTS)
3348
3349         elif n.type == token.COLONEQUAL:
3350             features.add(Feature.ASSIGNMENT_EXPRESSIONS)
3351
3352         elif (
3353             n.type in {syms.typedargslist, syms.arglist}
3354             and n.children
3355             and n.children[-1].type == token.COMMA
3356         ):
3357             if n.type == syms.typedargslist:
3358                 feature = Feature.TRAILING_COMMA_IN_DEF
3359             else:
3360                 feature = Feature.TRAILING_COMMA_IN_CALL
3361
3362             for ch in n.children:
3363                 if ch.type in STARS:
3364                     features.add(feature)
3365
3366                 if ch.type == syms.argument:
3367                     for argch in ch.children:
3368                         if argch.type in STARS:
3369                             features.add(feature)
3370
3371     return features
3372
3373
3374 def detect_target_versions(node: Node) -> Set[TargetVersion]:
3375     """Detect the version to target based on the nodes used."""
3376     features = get_features_used(node)
3377     return {
3378         version for version in TargetVersion if features <= VERSION_TO_FEATURES[version]
3379     }
3380
3381
3382 def generate_trailers_to_omit(line: Line, line_length: int) -> Iterator[Set[LeafID]]:
3383     """Generate sets of closing bracket IDs that should be omitted in a RHS.
3384
3385     Brackets can be omitted if the entire trailer up to and including
3386     a preceding closing bracket fits in one line.
3387
3388     Yielded sets are cumulative (contain results of previous yields, too).  First
3389     set is empty.
3390     """
3391
3392     omit: Set[LeafID] = set()
3393     yield omit
3394
3395     length = 4 * line.depth
3396     opening_bracket = None
3397     closing_bracket = None
3398     inner_brackets: Set[LeafID] = set()
3399     for index, leaf, leaf_length in enumerate_with_length(line, reversed=True):
3400         length += leaf_length
3401         if length > line_length:
3402             break
3403
3404         has_inline_comment = leaf_length > len(leaf.value) + len(leaf.prefix)
3405         if leaf.type == STANDALONE_COMMENT or has_inline_comment:
3406             break
3407
3408         if opening_bracket:
3409             if leaf is opening_bracket:
3410                 opening_bracket = None
3411             elif leaf.type in CLOSING_BRACKETS:
3412                 inner_brackets.add(id(leaf))
3413         elif leaf.type in CLOSING_BRACKETS:
3414             if index > 0 and line.leaves[index - 1].type in OPENING_BRACKETS:
3415                 # Empty brackets would fail a split so treat them as "inner"
3416                 # brackets (e.g. only add them to the `omit` set if another
3417                 # pair of brackets was good enough.
3418                 inner_brackets.add(id(leaf))
3419                 continue
3420
3421             if closing_bracket:
3422                 omit.add(id(closing_bracket))
3423                 omit.update(inner_brackets)
3424                 inner_brackets.clear()
3425                 yield omit
3426
3427             if leaf.value:
3428                 opening_bracket = leaf.opening_bracket
3429                 closing_bracket = leaf
3430
3431
3432 def get_future_imports(node: Node) -> Set[str]:
3433     """Return a set of __future__ imports in the file."""
3434     imports: Set[str] = set()
3435
3436     def get_imports_from_children(children: List[LN]) -> Generator[str, None, None]:
3437         for child in children:
3438             if isinstance(child, Leaf):
3439                 if child.type == token.NAME:
3440                     yield child.value
3441
3442             elif child.type == syms.import_as_name:
3443                 orig_name = child.children[0]
3444                 assert isinstance(orig_name, Leaf), "Invalid syntax parsing imports"
3445                 assert orig_name.type == token.NAME, "Invalid syntax parsing imports"
3446                 yield orig_name.value
3447
3448             elif child.type == syms.import_as_names:
3449                 yield from get_imports_from_children(child.children)
3450
3451             else:
3452                 raise AssertionError("Invalid syntax parsing imports")
3453
3454     for child in node.children:
3455         if child.type != syms.simple_stmt:
3456             break
3457
3458         first_child = child.children[0]
3459         if isinstance(first_child, Leaf):
3460             # Continue looking if we see a docstring; otherwise stop.
3461             if (
3462                 len(child.children) == 2
3463                 and first_child.type == token.STRING
3464                 and child.children[1].type == token.NEWLINE
3465             ):
3466                 continue
3467
3468             break
3469
3470         elif first_child.type == syms.import_from:
3471             module_name = first_child.children[1]
3472             if not isinstance(module_name, Leaf) or module_name.value != "__future__":
3473                 break
3474
3475             imports |= set(get_imports_from_children(first_child.children[3:]))
3476         else:
3477             break
3478
3479     return imports
3480
3481
3482 @lru_cache()
3483 def get_gitignore(root: Path) -> PathSpec:
3484     """ Return a PathSpec matching gitignore content if present."""
3485     gitignore = root / ".gitignore"
3486     lines: List[str] = []
3487     if gitignore.is_file():
3488         with gitignore.open() as gf:
3489             lines = gf.readlines()
3490     return PathSpec.from_lines("gitwildmatch", lines)
3491
3492
3493 def gen_python_files_in_dir(
3494     path: Path,
3495     root: Path,
3496     include: Pattern[str],
3497     exclude: Pattern[str],
3498     report: "Report",
3499     gitignore: PathSpec,
3500 ) -> Iterator[Path]:
3501     """Generate all files under `path` whose paths are not excluded by the
3502     `exclude` regex, but are included by the `include` regex.
3503
3504     Symbolic links pointing outside of the `root` directory are ignored.
3505
3506     `report` is where output about exclusions goes.
3507     """
3508     assert root.is_absolute(), f"INTERNAL ERROR: `root` must be absolute but is {root}"
3509     for child in path.iterdir():
3510         # First ignore files matching .gitignore
3511         if gitignore.match_file(child.as_posix()):
3512             report.path_ignored(child, f"matches the .gitignore file content")
3513             continue
3514
3515         # Then ignore with `exclude` option.
3516         try:
3517             normalized_path = "/" + child.resolve().relative_to(root).as_posix()
3518         except OSError as e:
3519             report.path_ignored(child, f"cannot be read because {e}")
3520             continue
3521
3522         except ValueError:
3523             if child.is_symlink():
3524                 report.path_ignored(
3525                     child, f"is a symbolic link that points outside {root}"
3526                 )
3527                 continue
3528
3529             raise
3530
3531         if child.is_dir():
3532             normalized_path += "/"
3533
3534         exclude_match = exclude.search(normalized_path)
3535         if exclude_match and exclude_match.group(0):
3536             report.path_ignored(child, f"matches the --exclude regular expression")
3537             continue
3538
3539         if child.is_dir():
3540             yield from gen_python_files_in_dir(
3541                 child, root, include, exclude, report, gitignore
3542             )
3543
3544         elif child.is_file():
3545             include_match = include.search(normalized_path)
3546             if include_match:
3547                 yield child
3548
3549
3550 @lru_cache()
3551 def find_project_root(srcs: Iterable[str]) -> Path:
3552     """Return a directory containing .git, .hg, or pyproject.toml.
3553
3554     That directory can be one of the directories passed in `srcs` or their
3555     common parent.
3556
3557     If no directory in the tree contains a marker that would specify it's the
3558     project root, the root of the file system is returned.
3559     """
3560     if not srcs:
3561         return Path("/").resolve()
3562
3563     common_base = min(Path(src).resolve() for src in srcs)
3564     if common_base.is_dir():
3565         # Append a fake file so `parents` below returns `common_base_dir`, too.
3566         common_base /= "fake-file"
3567     for directory in common_base.parents:
3568         if (directory / ".git").is_dir():
3569             return directory
3570
3571         if (directory / ".hg").is_dir():
3572             return directory
3573
3574         if (directory / "pyproject.toml").is_file():
3575             return directory
3576
3577     return directory
3578
3579
3580 @dataclass
3581 class Report:
3582     """Provides a reformatting counter. Can be rendered with `str(report)`."""
3583
3584     check: bool = False
3585     quiet: bool = False
3586     verbose: bool = False
3587     change_count: int = 0
3588     same_count: int = 0
3589     failure_count: int = 0
3590
3591     def done(self, src: Path, changed: Changed) -> None:
3592         """Increment the counter for successful reformatting. Write out a message."""
3593         if changed is Changed.YES:
3594             reformatted = "would reformat" if self.check else "reformatted"
3595             if self.verbose or not self.quiet:
3596                 out(f"{reformatted} {src}")
3597             self.change_count += 1
3598         else:
3599             if self.verbose:
3600                 if changed is Changed.NO:
3601                     msg = f"{src} already well formatted, good job."
3602                 else:
3603                     msg = f"{src} wasn't modified on disk since last run."
3604                 out(msg, bold=False)
3605             self.same_count += 1
3606
3607     def failed(self, src: Path, message: str) -> None:
3608         """Increment the counter for failed reformatting. Write out a message."""
3609         err(f"error: cannot format {src}: {message}")
3610         self.failure_count += 1
3611
3612     def path_ignored(self, path: Path, message: str) -> None:
3613         if self.verbose:
3614             out(f"{path} ignored: {message}", bold=False)
3615
3616     @property
3617     def return_code(self) -> int:
3618         """Return the exit code that the app should use.
3619
3620         This considers the current state of changed files and failures:
3621         - if there were any failures, return 123;
3622         - if any files were changed and --check is being used, return 1;
3623         - otherwise return 0.
3624         """
3625         # According to http://tldp.org/LDP/abs/html/exitcodes.html starting with
3626         # 126 we have special return codes reserved by the shell.
3627         if self.failure_count:
3628             return 123
3629
3630         elif self.change_count and self.check:
3631             return 1
3632
3633         return 0
3634
3635     def __str__(self) -> str:
3636         """Render a color report of the current state.
3637
3638         Use `click.unstyle` to remove colors.
3639         """
3640         if self.check:
3641             reformatted = "would be reformatted"
3642             unchanged = "would be left unchanged"
3643             failed = "would fail to reformat"
3644         else:
3645             reformatted = "reformatted"
3646             unchanged = "left unchanged"
3647             failed = "failed to reformat"
3648         report = []
3649         if self.change_count:
3650             s = "s" if self.change_count > 1 else ""
3651             report.append(
3652                 click.style(f"{self.change_count} file{s} {reformatted}", bold=True)
3653             )
3654         if self.same_count:
3655             s = "s" if self.same_count > 1 else ""
3656             report.append(f"{self.same_count} file{s} {unchanged}")
3657         if self.failure_count:
3658             s = "s" if self.failure_count > 1 else ""
3659             report.append(
3660                 click.style(f"{self.failure_count} file{s} {failed}", fg="red")
3661             )
3662         return ", ".join(report) + "."
3663
3664
3665 def parse_ast(src: str) -> Union[ast.AST, ast3.AST, ast27.AST]:
3666     filename = "<unknown>"
3667     if sys.version_info >= (3, 8):
3668         # TODO: support Python 4+ ;)
3669         for minor_version in range(sys.version_info[1], 4, -1):
3670             try:
3671                 return ast.parse(src, filename, feature_version=(3, minor_version))
3672             except SyntaxError:
3673                 continue
3674     else:
3675         for feature_version in (7, 6):
3676             try:
3677                 return ast3.parse(src, filename, feature_version=feature_version)
3678             except SyntaxError:
3679                 continue
3680
3681     return ast27.parse(src)
3682
3683
3684 def _fixup_ast_constants(
3685     node: Union[ast.AST, ast3.AST, ast27.AST]
3686 ) -> Union[ast.AST, ast3.AST, ast27.AST]:
3687     """Map ast nodes deprecated in 3.8 to Constant."""
3688     if isinstance(node, (ast.Str, ast3.Str, ast27.Str, ast.Bytes, ast3.Bytes)):
3689         return ast.Constant(value=node.s)
3690
3691     if isinstance(node, (ast.Num, ast3.Num, ast27.Num)):
3692         return ast.Constant(value=node.n)
3693
3694     if isinstance(node, (ast.NameConstant, ast3.NameConstant)):
3695         return ast.Constant(value=node.value)
3696
3697     return node
3698
3699
3700 def assert_equivalent(src: str, dst: str) -> None:
3701     """Raise AssertionError if `src` and `dst` aren't equivalent."""
3702
3703     def _v(node: Union[ast.AST, ast3.AST, ast27.AST], depth: int = 0) -> Iterator[str]:
3704         """Simple visitor generating strings to compare ASTs by content."""
3705
3706         node = _fixup_ast_constants(node)
3707
3708         yield f"{'  ' * depth}{node.__class__.__name__}("
3709
3710         for field in sorted(node._fields):
3711             # TypeIgnore has only one field 'lineno' which breaks this comparison
3712             type_ignore_classes = (ast3.TypeIgnore, ast27.TypeIgnore)
3713             if sys.version_info >= (3, 8):
3714                 type_ignore_classes += (ast.TypeIgnore,)
3715             if isinstance(node, type_ignore_classes):
3716                 break
3717
3718             try:
3719                 value = getattr(node, field)
3720             except AttributeError:
3721                 continue
3722
3723             yield f"{'  ' * (depth+1)}{field}="
3724
3725             if isinstance(value, list):
3726                 for item in value:
3727                     # Ignore nested tuples within del statements, because we may insert
3728                     # parentheses and they change the AST.
3729                     if (
3730                         field == "targets"
3731                         and isinstance(node, (ast.Delete, ast3.Delete, ast27.Delete))
3732                         and isinstance(item, (ast.Tuple, ast3.Tuple, ast27.Tuple))
3733                     ):
3734                         for item in item.elts:
3735                             yield from _v(item, depth + 2)
3736
3737                     elif isinstance(item, (ast.AST, ast3.AST, ast27.AST)):
3738                         yield from _v(item, depth + 2)
3739
3740             elif isinstance(value, (ast.AST, ast3.AST, ast27.AST)):
3741                 yield from _v(value, depth + 2)
3742
3743             else:
3744                 yield f"{'  ' * (depth+2)}{value!r},  # {value.__class__.__name__}"
3745
3746         yield f"{'  ' * depth})  # /{node.__class__.__name__}"
3747
3748     try:
3749         src_ast = parse_ast(src)
3750     except Exception as exc:
3751         raise AssertionError(
3752             f"cannot use --safe with this file; failed to parse source file.  "
3753             f"AST error message: {exc}"
3754         )
3755
3756     try:
3757         dst_ast = parse_ast(dst)
3758     except Exception as exc:
3759         log = dump_to_file("".join(traceback.format_tb(exc.__traceback__)), dst)
3760         raise AssertionError(
3761             f"INTERNAL ERROR: Black produced invalid code: {exc}. "
3762             f"Please report a bug on https://github.com/psf/black/issues.  "
3763             f"This invalid output might be helpful: {log}"
3764         ) from None
3765
3766     src_ast_str = "\n".join(_v(src_ast))
3767     dst_ast_str = "\n".join(_v(dst_ast))
3768     if src_ast_str != dst_ast_str:
3769         log = dump_to_file(diff(src_ast_str, dst_ast_str, "src", "dst"))
3770         raise AssertionError(
3771             f"INTERNAL ERROR: Black produced code that is not equivalent to "
3772             f"the source.  "
3773             f"Please report a bug on https://github.com/psf/black/issues.  "
3774             f"This diff might be helpful: {log}"
3775         ) from None
3776
3777
3778 def assert_stable(src: str, dst: str, mode: FileMode) -> None:
3779     """Raise AssertionError if `dst` reformats differently the second time."""
3780     newdst = format_str(dst, mode=mode)
3781     if dst != newdst:
3782         log = dump_to_file(
3783             diff(src, dst, "source", "first pass"),
3784             diff(dst, newdst, "first pass", "second pass"),
3785         )
3786         raise AssertionError(
3787             f"INTERNAL ERROR: Black produced different code on the second pass "
3788             f"of the formatter.  "
3789             f"Please report a bug on https://github.com/psf/black/issues.  "
3790             f"This diff might be helpful: {log}"
3791         ) from None
3792
3793
3794 def dump_to_file(*output: str) -> str:
3795     """Dump `output` to a temporary file. Return path to the file."""
3796     with tempfile.NamedTemporaryFile(
3797         mode="w", prefix="blk_", suffix=".log", delete=False, encoding="utf8"
3798     ) as f:
3799         for lines in output:
3800             f.write(lines)
3801             if lines and lines[-1] != "\n":
3802                 f.write("\n")
3803     return f.name
3804
3805
3806 @contextmanager
3807 def nullcontext() -> Iterator[None]:
3808     """Return an empty context manager.
3809
3810     To be used like `nullcontext` in Python 3.7.
3811     """
3812     yield
3813
3814
3815 def diff(a: str, b: str, a_name: str, b_name: str) -> str:
3816     """Return a unified diff string between strings `a` and `b`."""
3817     import difflib
3818
3819     a_lines = [line + "\n" for line in a.split("\n")]
3820     b_lines = [line + "\n" for line in b.split("\n")]
3821     return "".join(
3822         difflib.unified_diff(a_lines, b_lines, fromfile=a_name, tofile=b_name, n=5)
3823     )
3824
3825
3826 def cancel(tasks: Iterable[asyncio.Task]) -> None:
3827     """asyncio signal handler that cancels all `tasks` and reports to stderr."""
3828     err("Aborted!")
3829     for task in tasks:
3830         task.cancel()
3831
3832
3833 def shutdown(loop: asyncio.AbstractEventLoop) -> None:
3834     """Cancel all pending tasks on `loop`, wait for them, and close the loop."""
3835     try:
3836         if sys.version_info[:2] >= (3, 7):
3837             all_tasks = asyncio.all_tasks
3838         else:
3839             all_tasks = asyncio.Task.all_tasks
3840         # This part is borrowed from asyncio/runners.py in Python 3.7b2.
3841         to_cancel = [task for task in all_tasks(loop) if not task.done()]
3842         if not to_cancel:
3843             return
3844
3845         for task in to_cancel:
3846             task.cancel()
3847         loop.run_until_complete(
3848             asyncio.gather(*to_cancel, loop=loop, return_exceptions=True)
3849         )
3850     finally:
3851         # `concurrent.futures.Future` objects cannot be cancelled once they
3852         # are already running. There might be some when the `shutdown()` happened.
3853         # Silence their logger's spew about the event loop being closed.
3854         cf_logger = logging.getLogger("concurrent.futures")
3855         cf_logger.setLevel(logging.CRITICAL)
3856         loop.close()
3857
3858
3859 def sub_twice(regex: Pattern[str], replacement: str, original: str) -> str:
3860     """Replace `regex` with `replacement` twice on `original`.
3861
3862     This is used by string normalization to perform replaces on
3863     overlapping matches.
3864     """
3865     return regex.sub(replacement, regex.sub(replacement, original))
3866
3867
3868 def re_compile_maybe_verbose(regex: str) -> Pattern[str]:
3869     """Compile a regular expression string in `regex`.
3870
3871     If it contains newlines, use verbose mode.
3872     """
3873     if "\n" in regex:
3874         regex = "(?x)" + regex
3875     compiled: Pattern[str] = re.compile(regex)
3876     return compiled
3877
3878
3879 def enumerate_reversed(sequence: Sequence[T]) -> Iterator[Tuple[Index, T]]:
3880     """Like `reversed(enumerate(sequence))` if that were possible."""
3881     index = len(sequence) - 1
3882     for element in reversed(sequence):
3883         yield (index, element)
3884         index -= 1
3885
3886
3887 def enumerate_with_length(
3888     line: Line, reversed: bool = False
3889 ) -> Iterator[Tuple[Index, Leaf, int]]:
3890     """Return an enumeration of leaves with their length.
3891
3892     Stops prematurely on multiline strings and standalone comments.
3893     """
3894     op = cast(
3895         Callable[[Sequence[Leaf]], Iterator[Tuple[Index, Leaf]]],
3896         enumerate_reversed if reversed else enumerate,
3897     )
3898     for index, leaf in op(line.leaves):
3899         length = len(leaf.prefix) + len(leaf.value)
3900         if "\n" in leaf.value:
3901             return  # Multiline strings, we can't continue.
3902
3903         for comment in line.comments_after(leaf):
3904             length += len(comment.value)
3905
3906         yield index, leaf, length
3907
3908
3909 def is_line_short_enough(line: Line, *, line_length: int, line_str: str = "") -> bool:
3910     """Return True if `line` is no longer than `line_length`.
3911
3912     Uses the provided `line_str` rendering, if any, otherwise computes a new one.
3913     """
3914     if not line_str:
3915         line_str = str(line).strip("\n")
3916     return (
3917         len(line_str) <= line_length
3918         and "\n" not in line_str  # multiline strings
3919         and not line.contains_standalone_comments()
3920     )
3921
3922
3923 def can_be_split(line: Line) -> bool:
3924     """Return False if the line cannot be split *for sure*.
3925
3926     This is not an exhaustive search but a cheap heuristic that we can use to
3927     avoid some unfortunate formattings (mostly around wrapping unsplittable code
3928     in unnecessary parentheses).
3929     """
3930     leaves = line.leaves
3931     if len(leaves) < 2:
3932         return False
3933
3934     if leaves[0].type == token.STRING and leaves[1].type == token.DOT:
3935         call_count = 0
3936         dot_count = 0
3937         next = leaves[-1]
3938         for leaf in leaves[-2::-1]:
3939             if leaf.type in OPENING_BRACKETS:
3940                 if next.type not in CLOSING_BRACKETS:
3941                     return False
3942
3943                 call_count += 1
3944             elif leaf.type == token.DOT:
3945                 dot_count += 1
3946             elif leaf.type == token.NAME:
3947                 if not (next.type == token.DOT or next.type in OPENING_BRACKETS):
3948                     return False
3949
3950             elif leaf.type not in CLOSING_BRACKETS:
3951                 return False
3952
3953             if dot_count > 1 and call_count > 1:
3954                 return False
3955
3956     return True
3957
3958
3959 def can_omit_invisible_parens(line: Line, line_length: int) -> bool:
3960     """Does `line` have a shape safe to reformat without optional parens around it?
3961
3962     Returns True for only a subset of potentially nice looking formattings but
3963     the point is to not return false positives that end up producing lines that
3964     are too long.
3965     """
3966     bt = line.bracket_tracker
3967     if not bt.delimiters:
3968         # Without delimiters the optional parentheses are useless.
3969         return True
3970
3971     max_priority = bt.max_delimiter_priority()
3972     if bt.delimiter_count_with_priority(max_priority) > 1:
3973         # With more than one delimiter of a kind the optional parentheses read better.
3974         return False
3975
3976     if max_priority == DOT_PRIORITY:
3977         # A single stranded method call doesn't require optional parentheses.
3978         return True
3979
3980     assert len(line.leaves) >= 2, "Stranded delimiter"
3981
3982     first = line.leaves[0]
3983     second = line.leaves[1]
3984     penultimate = line.leaves[-2]
3985     last = line.leaves[-1]
3986
3987     # With a single delimiter, omit if the expression starts or ends with
3988     # a bracket.
3989     if first.type in OPENING_BRACKETS and second.type not in CLOSING_BRACKETS:
3990         remainder = False
3991         length = 4 * line.depth
3992         for _index, leaf, leaf_length in enumerate_with_length(line):
3993             if leaf.type in CLOSING_BRACKETS and leaf.opening_bracket is first:
3994                 remainder = True
3995             if remainder:
3996                 length += leaf_length
3997                 if length > line_length:
3998                     break
3999
4000                 if leaf.type in OPENING_BRACKETS:
4001                     # There are brackets we can further split on.
4002                     remainder = False
4003
4004         else:
4005             # checked the entire string and line length wasn't exceeded
4006             if len(line.leaves) == _index + 1:
4007                 return True
4008
4009         # Note: we are not returning False here because a line might have *both*
4010         # a leading opening bracket and a trailing closing bracket.  If the
4011         # opening bracket doesn't match our rule, maybe the closing will.
4012
4013     if (
4014         last.type == token.RPAR
4015         or last.type == token.RBRACE
4016         or (
4017             # don't use indexing for omitting optional parentheses;
4018             # it looks weird
4019             last.type == token.RSQB
4020             and last.parent
4021             and last.parent.type != syms.trailer
4022         )
4023     ):
4024         if penultimate.type in OPENING_BRACKETS:
4025             # Empty brackets don't help.
4026             return False
4027
4028         if is_multiline_string(first):
4029             # Additional wrapping of a multiline string in this situation is
4030             # unnecessary.
4031             return True
4032
4033         length = 4 * line.depth
4034         seen_other_brackets = False
4035         for _index, leaf, leaf_length in enumerate_with_length(line):
4036             length += leaf_length
4037             if leaf is last.opening_bracket:
4038                 if seen_other_brackets or length <= line_length:
4039                     return True
4040
4041             elif leaf.type in OPENING_BRACKETS:
4042                 # There are brackets we can further split on.
4043                 seen_other_brackets = True
4044
4045     return False
4046
4047
4048 def get_cache_file(mode: FileMode) -> Path:
4049     return CACHE_DIR / f"cache.{mode.get_cache_key()}.pickle"
4050
4051
4052 def read_cache(mode: FileMode) -> Cache:
4053     """Read the cache if it exists and is well formed.
4054
4055     If it is not well formed, the call to write_cache later should resolve the issue.
4056     """
4057     cache_file = get_cache_file(mode)
4058     if not cache_file.exists():
4059         return {}
4060
4061     with cache_file.open("rb") as fobj:
4062         try:
4063             cache: Cache = pickle.load(fobj)
4064         except (pickle.UnpicklingError, ValueError):
4065             return {}
4066
4067     return cache
4068
4069
4070 def get_cache_info(path: Path) -> CacheInfo:
4071     """Return the information used to check if a file is already formatted or not."""
4072     stat = path.stat()
4073     return stat.st_mtime, stat.st_size
4074
4075
4076 def filter_cached(cache: Cache, sources: Iterable[Path]) -> Tuple[Set[Path], Set[Path]]:
4077     """Split an iterable of paths in `sources` into two sets.
4078
4079     The first contains paths of files that modified on disk or are not in the
4080     cache. The other contains paths to non-modified files.
4081     """
4082     todo, done = set(), set()
4083     for src in sources:
4084         src = src.resolve()
4085         if cache.get(src) != get_cache_info(src):
4086             todo.add(src)
4087         else:
4088             done.add(src)
4089     return todo, done
4090
4091
4092 def write_cache(cache: Cache, sources: Iterable[Path], mode: FileMode) -> None:
4093     """Update the cache file."""
4094     cache_file = get_cache_file(mode)
4095     try:
4096         CACHE_DIR.mkdir(parents=True, exist_ok=True)
4097         new_cache = {**cache, **{src.resolve(): get_cache_info(src) for src in sources}}
4098         with tempfile.NamedTemporaryFile(dir=str(cache_file.parent), delete=False) as f:
4099             pickle.dump(new_cache, f, protocol=4)
4100         os.replace(f.name, cache_file)
4101     except OSError:
4102         pass
4103
4104
4105 def patch_click() -> None:
4106     """Make Click not crash.
4107
4108     On certain misconfigured environments, Python 3 selects the ASCII encoding as the
4109     default which restricts paths that it can access during the lifetime of the
4110     application.  Click refuses to work in this scenario by raising a RuntimeError.
4111
4112     In case of Black the likelihood that non-ASCII characters are going to be used in
4113     file paths is minimal since it's Python source code.  Moreover, this crash was
4114     spurious on Python 3.7 thanks to PEP 538 and PEP 540.
4115     """
4116     try:
4117         from click import core
4118         from click import _unicodefun  # type: ignore
4119     except ModuleNotFoundError:
4120         return
4121
4122     for module in (core, _unicodefun):
4123         if hasattr(module, "_verify_python3_env"):
4124             module._verify_python3_env = lambda: None
4125
4126
4127 def patched_main() -> None:
4128     freeze_support()
4129     patch_click()
4130     main()
4131
4132
4133 if __name__ == "__main__":
4134     patched_main()