return
elif len(sources) == 1:
- return_code = run_single_file_mode(
- line_length, check, fast, quiet, write_back, sources[0]
- )
+ return_code = reformat_one(sources[0], line_length, fast, quiet, write_back)
else:
- return_code = run_multi_file_mode(line_length, fast, quiet, write_back, sources)
+ loop = asyncio.get_event_loop()
+ executor = ProcessPoolExecutor(max_workers=os.cpu_count())
+ return_code = 1
+ try:
+ return_code = loop.run_until_complete(
+ schedule_formatting(
+ sources, line_length, write_back, fast, quiet, loop, executor
+ )
+ )
+ finally:
+ shutdown(loop)
ctx.exit(return_code)
-def run_single_file_mode(
- line_length: int,
- check: bool,
- fast: bool,
- quiet: bool,
- write_back: WriteBack,
- src: Path,
+def reformat_one(
+ src: Path, line_length: int, fast: bool, quiet: bool, write_back: WriteBack
) -> int:
- report = Report(check=check, quiet=quiet)
+ """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) == "-":
return report.return_code
-def run_multi_file_mode(
- line_length: int,
- fast: bool,
- quiet: bool,
- write_back: WriteBack,
- sources: List[Path],
-) -> int:
- loop = asyncio.get_event_loop()
- executor = ProcessPoolExecutor(max_workers=os.cpu_count())
- return_code = 1
- try:
- return_code = loop.run_until_complete(
- schedule_formatting(
- sources, line_length, write_back, fast, quiet, loop, executor
- )
- )
- finally:
- shutdown(loop)
- return return_code
-
-
async def schedule_formatting(
sources: List[Path],
line_length: int,
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)
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 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.
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