Union,
cast,
)
+from typing_extensions import Final
+from mypy_extensions import mypyc_attr
from appdirs import user_cache_dir
from dataclasses import dataclass, field, replace
return value
+def target_version_option_callback(
+ c: click.Context, p: Union[click.Option, click.Parameter], v: Tuple[str, ...]
+) -> List[TargetVersion]:
+ """Compute the target versions from a --target-version flag.
+
+ This is its own function because mypy couldn't infer the type correctly
+ when it was a lambda, causing mypyc trouble.
+ """
+ return [TargetVersion[val.upper()] for val in v]
+
+
@click.command(context_settings=dict(help_option_names=["-h", "--help"]))
@click.option("-c", "--code", type=str, help="Format the code passed in as a string.")
@click.option(
"-t",
"--target-version",
type=click.Choice([v.name.lower() for v in TargetVersion]),
- callback=lambda c, p, v: [TargetVersion[val.upper()] for val in v],
+ callback=target_version_option_callback,
multiple=True,
help=(
"Python versions that should be supported by Black's output. [default: "
verbose: bool,
include: str,
exclude: str,
- src: Tuple[str],
+ src: Tuple[str, ...],
config: Optional[str],
) -> None:
"""The uncompromising code formatter."""
except re.error:
err(f"Invalid regular expression for exclude given: {exclude!r}")
ctx.exit(2)
- report = Report(check=check, quiet=quiet, verbose=verbose)
+ report = Report(check=check, diff=diff, quiet=quiet, verbose=verbose)
root = find_project_root(src)
sources: Set[Path] = set()
path_empty(src, quiet, verbose, ctx)
ctx.exit(report.return_code)
-def path_empty(src: Tuple[str], quiet: bool, verbose: bool, ctx: click.Context) -> None:
+def path_empty(
+ src: Tuple[str, ...], quiet: bool, verbose: bool, ctx: click.Context
+) -> None:
"""
Exit if there is no `src` provided for formatting
"""
): src
for src in sorted(sources)
}
- pending: Iterable[asyncio.Future] = tasks.keys()
+ pending: Iterable["asyncio.Future[bool]"] = tasks.keys()
try:
loop.add_signal_handler(signal.SIGINT, cancel, pending)
loop.add_signal_handler(signal.SIGTERM, cancel, pending)
except NothingChanged:
return False
- if write_back == write_back.YES:
+ if write_back == WriteBack.YES:
with open(src, "w", encoding=encoding, newline=newline) as f:
f.write(dst_contents)
- elif write_back == write_back.DIFF:
+ elif write_back == WriteBack.DIFF:
now = datetime.utcnow()
src_name = f"{src}\t{then} +0000"
dst_name = f"{src}\t{now} +0000"
if supports_feature(versions, feature)
}
for current_line in lines.visit(src_node):
- for _ in range(after):
- dst_contents.append(str(empty_line))
+ dst_contents.append(str(empty_line) * after)
before, after = elt.maybe_empty_lines(current_line)
- for _ in range(before):
- dst_contents.append(str(empty_line))
+ dst_contents.append(str(empty_line) * before)
for line in split_line(
current_line, line_length=mode.line_length, features=split_line_features
):
if node.type < 256:
name = token.tok_name[node.type]
else:
- name = type_repr(node.type)
- yield from getattr(self, f"visit_{name}", self.visit_default)(node)
+ name = str(type_repr(node.type))
+ # We explicitly branch on whether a visitor exists (instead of
+ # using self.visit_default as the default arg to getattr) in order
+ # to save needing to create a bound method object and so mypyc can
+ # generate a native call to visit_default.
+ visitf = getattr(self, f"visit_{name}", None)
+ if visitf:
+ yield from visitf(node)
+ else:
+ yield from self.visit_default(node)
def visit_default(self, node: LN) -> Iterator[T]:
"""Default `visit_*()` implementation. Recurses to children of `node`."""
list(v.visit(code))
-WHITESPACE = {token.DEDENT, token.INDENT, token.NEWLINE}
-STATEMENT = {
+WHITESPACE: Final = {token.DEDENT, token.INDENT, token.NEWLINE}
+STATEMENT: Final = {
syms.if_stmt,
syms.while_stmt,
syms.for_stmt,
syms.funcdef,
syms.classdef,
}
-STANDALONE_COMMENT = 153
+STANDALONE_COMMENT: Final = 153
token.tok_name[STANDALONE_COMMENT] = "STANDALONE_COMMENT"
-LOGIC_OPERATORS = {"and", "or"}
-COMPARATORS = {
+LOGIC_OPERATORS: Final = {"and", "or"}
+COMPARATORS: Final = {
token.LESS,
token.GREATER,
token.EQEQUAL,
token.LESSEQUAL,
token.GREATEREQUAL,
}
-MATH_OPERATORS = {
+MATH_OPERATORS: Final = {
token.VBAR,
token.CIRCUMFLEX,
token.AMPER,
token.TILDE,
token.DOUBLESTAR,
}
-STARS = {token.STAR, token.DOUBLESTAR}
-VARARGS_SPECIALS = STARS | {token.SLASH}
-VARARGS_PARENTS = {
+STARS: Final = {token.STAR, token.DOUBLESTAR}
+VARARGS_SPECIALS: Final = STARS | {token.SLASH}
+VARARGS_PARENTS: Final = {
syms.arglist,
syms.argument, # double star in arglist
syms.trailer, # single argument to call
syms.typedargslist,
syms.varargslist, # lambdas
}
-UNPACKING_PARENTS = {
+UNPACKING_PARENTS: Final = {
syms.atom, # single element of a list or set literal
syms.dictsetmaker,
syms.listmaker,
syms.testlist_gexp,
syms.testlist_star_expr,
}
-TEST_DESCENDANTS = {
+TEST_DESCENDANTS: Final = {
syms.test,
syms.lambdef,
syms.or_test,
syms.term,
syms.power,
}
-ASSIGNMENTS = {
+ASSIGNMENTS: Final = {
"=",
"+=",
"-=",
"**=",
"//=",
}
-COMPREHENSION_PRIORITY = 20
-COMMA_PRIORITY = 18
-TERNARY_PRIORITY = 16
-LOGIC_PRIORITY = 14
-STRING_PRIORITY = 12
-COMPARATOR_PRIORITY = 10
-MATH_PRIORITIES = {
+COMPREHENSION_PRIORITY: Final = 20
+COMMA_PRIORITY: Final = 18
+TERNARY_PRIORITY: Final = 16
+LOGIC_PRIORITY: Final = 14
+STRING_PRIORITY: Final = 12
+COMPARATOR_PRIORITY: Final = 10
+MATH_PRIORITIES: Final = {
token.VBAR: 9,
token.CIRCUMFLEX: 8,
token.AMPER: 7,
token.TILDE: 3,
token.DOUBLESTAR: 2,
}
-DOT_PRIORITY = 1
+DOT_PRIORITY: Final = 1
@dataclass
for leaf_id, comments in self.comments.items():
for comment in comments:
if is_type_comment(comment):
- if leaf_id not in ignored_ids or comment_seen:
+ if comment_seen or (
+ not is_type_comment(comment, " ignore")
+ and leaf_id not in ignored_ids
+ ):
return True
comment_seen = True
return False
def contains_multiline_strings(self) -> bool:
- for leaf in self.leaves:
- if is_multiline_string(leaf):
- return True
-
- return False
+ return any(is_multiline_string(leaf) for leaf in self.leaves)
def maybe_remove_trailing_comma(self, closing: Leaf) -> bool:
"""Remove trailing comma if there is one and it's safe."""
self.current_line.append(node)
yield from super().visit_default(node)
- def visit_INDENT(self, node: Node) -> Iterator[Line]:
+ def visit_INDENT(self, node: Leaf) -> Iterator[Line]:
"""Increase indentation level, maybe yield a line."""
# In blib2to3 INDENT never holds comments.
yield from self.line(+1)
yield from self.visit_default(node)
- def visit_DEDENT(self, node: Node) -> Iterator[Line]:
+ def visit_DEDENT(self, node: Leaf) -> Iterator[Line]:
"""Decrease indentation level, maybe yield a line."""
# 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).
body_leaves: List[Leaf] = []
head_leaves: List[Leaf] = []
current_leaves = head_leaves
- matching_bracket = None
+ matching_bracket: Optional[Leaf] = None
for leaf in line.leaves:
if (
current_leaves is body_leaves
body_leaves: List[Leaf] = []
head_leaves: List[Leaf] = []
current_leaves = tail_leaves
- opening_bracket = None
- closing_bracket = None
+ opening_bracket: Optional[Leaf] = None
+ closing_bracket: Optional[Leaf] = None
for leaf in reversed(line.leaves):
if current_leaves is body_leaves:
if leaf is opening_bracket:
# That happens when one of the `ignored_nodes` ended with a NEWLINE
# leaf (possibly followed by a DEDENT).
hidden_value = hidden_value[:-1]
- first_idx = None
+ first_idx: Optional[int] = None
for ignored in ignored_nodes:
index = ignored.remove()
if first_idx is None:
yield omit
length = 4 * line.depth
- opening_bracket = None
- closing_bracket = None
+ opening_bracket: Optional[Leaf] = None
+ closing_bracket: Optional[Leaf] = None
inner_brackets: Set[LeafID] = set()
for index, leaf, leaf_length in enumerate_with_length(line, reversed=True):
length += leaf_length
# Append a fake file so `parents` below returns `common_base_dir`, too.
common_base /= "fake-file"
for directory in common_base.parents:
- if (directory / ".git").is_dir():
+ if (directory / ".git").exists():
return directory
if (directory / ".hg").is_dir():
"""Provides a reformatting counter. Can be rendered with `str(report)`."""
check: bool = False
+ diff: bool = False
quiet: bool = False
verbose: bool = False
change_count: int = 0
def done(self, src: Path, changed: Changed) -> None:
"""Increment the counter for successful reformatting. Write out a message."""
if changed is Changed.YES:
- reformatted = "would reformat" if self.check else "reformatted"
+ reformatted = "would reformat" if self.check or self.diff else "reformatted"
if self.verbose or not self.quiet:
out(f"{reformatted} {src}")
self.change_count += 1
Use `click.unstyle` to remove colors.
"""
- if self.check:
+ if self.check or self.diff:
reformatted = "would be reformatted"
unchanged = "would be left unchanged"
failed = "would fail to reformat"
) from None
+@mypyc_attr(patchable=True)
def dump_to_file(*output: str) -> str:
"""Dump `output` to a temporary file. Return path to the file."""
with tempfile.NamedTemporaryFile(
)
-def cancel(tasks: Iterable[asyncio.Task]) -> None:
+def cancel(tasks: Iterable["asyncio.Task[Any]"]) -> None:
"""asyncio signal handler that cancels all `tasks` and reports to stderr."""
err("Aborted!")
for task in tasks: