from blib2to3.pgen2 import driver, token
from blib2to3.pgen2.parse import ParseError
-__version__ = "18.4a2"
+__version__ = "18.4a4"
DEFAULT_LINE_LENGTH = 88
+
# types
syms = pygram.python_symbols
FileContent = str
sources.append(Path("-"))
else:
err(f"invalid path: {s}")
- if check and diff:
- exc = click.ClickException("Options --check and --diff are mutually exclusive")
- exc.exit_code = 2
- raise exc
- if check:
+ if check and not diff:
write_back = WriteBack.NO
elif diff:
write_back = WriteBack.DIFF
else:
write_back = WriteBack.YES
+ report = Report(check=check, quiet=quiet)
if len(sources) == 0:
ctx.exit(0)
return
elif len(sources) == 1:
- return_code = reformat_one(sources[0], line_length, fast, quiet, write_back)
+ reformat_one(sources[0], line_length, fast, write_back, report)
else:
loop = asyncio.get_event_loop()
executor = ProcessPoolExecutor(max_workers=os.cpu_count())
- return_code = 1
try:
- return_code = loop.run_until_complete(
+ loop.run_until_complete(
schedule_formatting(
- sources, line_length, write_back, fast, quiet, loop, executor
+ sources, line_length, fast, write_back, report, loop, executor
)
)
finally:
shutdown(loop)
- ctx.exit(return_code)
+ if not quiet:
+ out("All done! ✨ 🍰 ✨")
+ click.echo(str(report))
+ ctx.exit(report.return_code)
def reformat_one(
- src: Path, line_length: int, fast: bool, quiet: bool, write_back: WriteBack
-) -> int:
+ src: Path, line_length: int, fast: bool, write_back: WriteBack, report: "Report"
+) -> None:
"""Reformat a single file under `src` without spawning child processes.
If `quiet` is True, non-error messages are not output. `line_length`,
`write_back`, and `fast` options are passed to :func:`format_file_in_place`.
"""
- report = Report(check=write_back is WriteBack.NO, quiet=quiet)
try:
changed = Changed.NO
if not src.is_file() and str(src) == "-":
else:
cache: Cache = {}
if write_back != WriteBack.DIFF:
- cache = read_cache()
+ cache = read_cache(line_length)
src = src.resolve()
if src in cache and cache[src] == get_cache_info(src):
changed = Changed.CACHED
)
):
changed = Changed.YES
- if write_back != WriteBack.DIFF and changed is not Changed.NO:
- write_cache(cache, [src])
+ if write_back == WriteBack.YES and changed is not Changed.NO:
+ write_cache(cache, [src], line_length)
report.done(src, changed)
except Exception as exc:
report.failed(src, str(exc))
- return report.return_code
async def schedule_formatting(
sources: List[Path],
line_length: int,
- write_back: WriteBack,
fast: bool,
- quiet: bool,
+ write_back: WriteBack,
+ report: "Report",
loop: BaseEventLoop,
executor: Executor,
-) -> int:
+) -> None:
"""Run formatting of `sources` in parallel using the provided `executor`.
(Use ProcessPoolExecutors for actual parallelism.)
`line_length`, `write_back`, and `fast` options are passed to
:func:`format_file_in_place`.
"""
- report = Report(check=write_back is WriteBack.NO, quiet=quiet)
cache: Cache = {}
if write_back != WriteBack.DIFF:
- cache = read_cache()
+ cache = read_cache(line_length)
sources, cached = filter_cached(cache, sources)
for src in cached:
report.done(src, Changed.CACHED)
for src in sources
}
_task_values = list(tasks.values())
- loop.add_signal_handler(signal.SIGINT, cancel, _task_values)
- loop.add_signal_handler(signal.SIGTERM, cancel, _task_values)
+ try:
+ loop.add_signal_handler(signal.SIGINT, cancel, _task_values)
+ loop.add_signal_handler(signal.SIGTERM, cancel, _task_values)
+ except NotImplementedError:
+ # There are no good alternatives for these on Windows
+ pass
await asyncio.wait(_task_values)
for src, task in tasks.items():
if not task.done():
if cancelled:
await asyncio.gather(*cancelled, loop=loop, return_exceptions=True)
- elif not quiet:
- out("All done! ✨ 🍰 ✨")
- if not quiet:
- click.echo(str(report))
-
- if write_back != WriteBack.DIFF and formatted:
- write_cache(cache, formatted)
-
- return report.return_code
+ if write_back == WriteBack.YES and formatted:
+ write_cache(cache, formatted, line_length)
def format_file_in_place(
with open(src, "w", encoding=src_buffer.encoding) as f:
f.write(dst_contents)
elif write_back == write_back.DIFF:
- src_name = f"{src.name} (original)"
- dst_name = f"{src.name} (formatted)"
+ src_name = f"{src} (original)"
+ dst_name = f"{src} (formatted)"
diff_contents = diff(src_contents, dst_contents, src_name, dst_name)
if lock:
lock.acquire()
GRAMMARS = [
pygram.python_grammar_no_print_statement_no_exec_statement,
pygram.python_grammar_no_print_statement,
- pygram.python_grammar_no_exec_statement,
pygram.python_grammar,
]
}
COMPREHENSION_PRIORITY = 20
COMMA_PRIORITY = 10
+TERNARY_PRIORITY = 7
LOGIC_PRIORITY = 5
STRING_PRIORITY = 4
COMPARATOR_PRIORITY = 3
bracket_match: Dict[Tuple[Depth, NodeType], Leaf] = Factory(dict)
delimiters: Dict[LeafID, Priority] = Factory(dict)
previous: Optional[Leaf] = None
+ _for_loop_variable: bool = False
+ _lambda_arguments: bool = False
def mark(self, leaf: Leaf) -> None:
"""Mark `leaf` with bracket-related metadata. Keep track of delimiters.
if leaf.type == token.COMMENT:
return
+ self.maybe_decrement_after_for_loop_variable(leaf)
+ self.maybe_decrement_after_lambda_arguments(leaf)
if leaf.type in CLOSING_BRACKETS:
self.depth -= 1
opening_bracket = self.bracket_match.pop((self.depth, leaf.type))
self.bracket_match[self.depth, BRACKET[leaf.type]] = leaf
self.depth += 1
self.previous = leaf
+ self.maybe_increment_lambda_arguments(leaf)
+ self.maybe_increment_for_loop_variable(leaf)
def any_open_brackets(self) -> bool:
"""Return True if there is an yet unmatched open bracket on the line."""
def max_delimiter_priority(self, exclude: Iterable[LeafID] = ()) -> int:
"""Return the highest priority of a delimiter found on the line.
- Values are consistent with what `is_delimiter()` returns.
+ Values are consistent with what `is_split_*_delimiter()` return.
Raises ValueError on no delimiters.
"""
return max(v for k, v in self.delimiters.items() if k not in exclude)
+ def maybe_increment_for_loop_variable(self, leaf: Leaf) -> bool:
+ """In a for loop, or comprehension, the variables are often unpacks.
+
+ To avoid splitting on the comma in this situation, increase the depth of
+ tokens between `for` and `in`.
+ """
+ if leaf.type == token.NAME and leaf.value == "for":
+ self.depth += 1
+ self._for_loop_variable = True
+ return True
+
+ return False
+
+ def maybe_decrement_after_for_loop_variable(self, leaf: Leaf) -> bool:
+ """See `maybe_increment_for_loop_variable` above for explanation."""
+ if self._for_loop_variable and leaf.type == token.NAME and leaf.value == "in":
+ self.depth -= 1
+ self._for_loop_variable = False
+ return True
+
+ return False
+
+ def maybe_increment_lambda_arguments(self, leaf: Leaf) -> bool:
+ """In a lambda expression, there might be more than one argument.
+
+ To avoid splitting on the comma in this situation, increase the depth of
+ tokens between `lambda` and `:`.
+ """
+ if leaf.type == token.NAME and leaf.value == "lambda":
+ self.depth += 1
+ self._lambda_arguments = True
+ return True
+
+ return False
+
+ def maybe_decrement_after_lambda_arguments(self, leaf: Leaf) -> bool:
+ """See `maybe_increment_lambda_arguments` above for explanation."""
+ if self._lambda_arguments and leaf.type == token.COLON:
+ self.depth -= 1
+ self._lambda_arguments = False
+ return True
+
+ return False
+
@dataclass
class Line:
comments: List[Tuple[Index, Leaf]] = Factory(list)
bracket_tracker: BracketTracker = Factory(BracketTracker)
inside_brackets: bool = False
- has_for: bool = False
- _for_loop_variable: bool = False
def append(self, leaf: Leaf, preformatted: bool = False) -> None:
"""Add a new `leaf` to the end of the line.
# imports, for which we only preserve newlines.
leaf.prefix += whitespace(leaf)
if self.inside_brackets or not preformatted:
- self.maybe_decrement_after_for_loop_variable(leaf)
self.bracket_tracker.mark(leaf)
self.maybe_remove_trailing_comma(leaf)
- self.maybe_increment_for_loop_variable(leaf)
if not self.append_comment(leaf):
self.leaves.append(leaf)
return False
- def maybe_increment_for_loop_variable(self, leaf: Leaf) -> bool:
- """In a for loop, or comprehension, the variables are often unpacks.
-
- To avoid splitting on the comma in this situation, increase the depth of
- tokens between `for` and `in`.
- """
- if leaf.type == token.NAME and leaf.value == "for":
- self.has_for = True
- self.bracket_tracker.depth += 1
- self._for_loop_variable = True
- return True
-
- return False
-
- def maybe_decrement_after_for_loop_variable(self, leaf: Leaf) -> bool:
- """See `maybe_increment_for_loop_variable` above for explanation."""
- if self._for_loop_variable and leaf.type == token.NAME and leaf.value == "in":
- self.bracket_tracker.depth -= 1
- self._for_loop_variable = False
- return True
-
- return False
-
def append_comment(self, comment: Leaf) -> bool:
"""Add an inline or standalone comment to the line."""
if (
# Don't insert empty lines before the first line in the file.
return 0, 0
- if self.previous_line and self.previous_line.is_decorator:
- # Don't insert empty lines between decorators.
+ if self.previous_line.is_decorator:
+ return 0, 0
+
+ if (
+ self.previous_line.is_comment
+ and self.previous_line.depth == current_line.depth
+ and before == 0
+ ):
return 0, 0
newlines = 2
newlines -= 1
return newlines, 0
- if current_line.is_flow_control:
- return before, 1
-
if (
self.previous_line
and self.previous_line.is_import
):
return (before or 1), 0
- if (
- self.previous_line
- and self.previous_line.is_yield
- and (not current_line.is_yield or depth != self.previous_line.depth)
- ):
- return (before or 1), 0
-
return before, 0
def visit_DEDENT(self, node: Node) -> Iterator[Line]:
"""Decrease indentation level, maybe yield a line."""
- # DEDENT has no value. Additionally, in blib2to3 it never holds comments.
+ # The current line might still wait for trailing comments. At DEDENT time
+ # there won't be any (they would be prefixes on the preceding NEWLINE).
+ # Emit the line then.
+ yield from self.line()
+
+ # While DEDENT has no value, its prefix may contain standalone comments
+ # that belong to the current indentation level. Get 'em.
+ yield from self.visit_default(node)
+
+ # Finally, emit the dedent.
yield from self.line(-1)
def visit_stmt(
if p.type in {syms.parameters, syms.arglist}:
# untyped function signatures or calls
- if t == token.RPAR:
- return NO
-
if not prev or prev.type != token.COMMA:
return NO
elif p.type == syms.varargslist:
# lambdas
- if t == token.RPAR:
- return NO
-
if prev and prev.type != token.COMMA:
return NO
# dots, but not the first one.
return NO
- elif (
- p.type == syms.listmaker
- or p.type == syms.testlist_gexp
- or p.type == syms.subscriptlist
- ):
- # list interior, including unpacking
- if not prev:
- return NO
-
elif p.type == syms.dictsetmaker:
- # dict and set interior, including unpacking
- if not prev:
- return NO
-
- if prev.type == token.DOUBLESTAR:
+ # dict unpacking
+ if prev and prev.type == token.DOUBLESTAR:
return NO
elif p.type in {syms.factor, syms.star_expr}:
):
return COMPREHENSION_PRIORITY
+ if (
+ leaf.type == token.NAME
+ and leaf.value in {"if", "else"}
+ and leaf.parent
+ and leaf.parent.type == syms.test
+ ):
+ return TERNARY_PRIORITY
+
if leaf.type == token.NAME and leaf.value in LOGIC_OPERATORS and leaf.parent:
return LOGIC_PRIORITY
return 0
-def is_delimiter(leaf: Leaf, previous: Leaf = None) -> int:
- """Return the priority of the `leaf` delimiter. Return 0 if not delimiter.
-
- Higher numbers are higher priority.
- """
- return max(
- is_split_before_delimiter(leaf, previous),
- is_split_after_delimiter(leaf, previous),
- )
-
-
def generate_comments(leaf: Leaf) -> Iterator[Leaf]:
"""Clean the prefix of the `leaf` and generate comments from it, if any.
split_funcs: List[SplitFunc]
if line.is_def:
split_funcs = [left_hand_split]
+ elif line.is_import:
+ split_funcs = [explode_split]
elif line.inside_brackets:
split_funcs = [delimiter_split, standalone_comment_split, right_hand_split]
else:
yield current_line
+def explode_split(
+ line: Line, py36: bool = False, omit: Collection[LeafID] = ()
+) -> Iterator[Line]:
+ """Split by rightmost bracket and immediately split contents by a delimiter."""
+ new_lines = list(right_hand_split(line, py36, omit))
+ if len(new_lines) != 3:
+ yield from new_lines
+ return
+
+ yield new_lines[0]
+
+ try:
+ yield from delimiter_split(new_lines[1], py36)
+ except CannotSplit:
+ yield new_lines[1]
+
+ yield new_lines[2]
+
+
def is_import(leaf: Leaf) -> bool:
"""Return True if the given leaf starts an import statement."""
p = leaf.parent
def max_delimiter_priority_in_atom(node: LN) -> int:
+ """Return maximum delimiter priority inside `node`.
+
+ This is specific to atoms with contents contained in a pair of parentheses.
+ If `node` isn't an atom or there are no enclosing parentheses, returns 0.
+ """
if node.type != syms.atom:
return 0
CACHE_DIR = Path(user_cache_dir("black", version=__version__))
-CACHE_FILE = CACHE_DIR / "cache.pickle"
-def read_cache() -> Cache:
+def get_cache_file(line_length: int) -> Path:
+ return CACHE_DIR / f"cache.{line_length}.pickle"
+
+
+def read_cache(line_length: int) -> Cache:
"""Read the cache if it exists and is well formed.
If it is not well formed, the call to write_cache later should resolve the issue.
"""
- if not CACHE_FILE.exists():
+ cache_file = get_cache_file(line_length)
+ if not cache_file.exists():
return {}
- with CACHE_FILE.open("rb") as fobj:
+ with cache_file.open("rb") as fobj:
try:
cache: Cache = pickle.load(fobj)
except pickle.UnpicklingError:
return todo, done
-def write_cache(cache: Cache, sources: List[Path]) -> None:
+def write_cache(cache: Cache, sources: List[Path], line_length: int) -> None:
"""Update the cache file."""
+ cache_file = get_cache_file(line_length)
try:
if not CACHE_DIR.exists():
CACHE_DIR.mkdir(parents=True)
new_cache = {**cache, **{src.resolve(): get_cache_info(src) for src in sources}}
- with CACHE_FILE.open("wb") as fobj:
+ with cache_file.open("wb") as fobj:
pickle.dump(new_cache, fobj, protocol=pickle.HIGHEST_PROTOCOL)
except OSError:
pass
projects / etc / vim.git / blobdiff