Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
except NotImplementedError:
# There are no good alternatives for these on Windows.
pass
while pending:
done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
for task in done:
src = tasks.pop(task)
if task.cancelled():
cancelled.append(task)
elif task.exception():
report.failed(src, str(task.exception()))
else:
changed = Changed.YES if task.result() else Changed.NO
# If the file was written back or was successfully checked as
# well-formatted, store this information in the cache.
if write_back is WriteBack.YES or (
write_back is WriteBack.CHECK and changed is Changed.NO
):
sources_to_cache.append(src)
report.done(src, changed)
if cancelled:
await asyncio.gather(*cancelled, loop=loop, return_exceptions=True)
if sources_to_cache:
write_cache(cache, sources_to_cache, mode)
def cli(
ctx: click.Context,
line_length: int,
check: bool,
include: str,
exclude: str,
quiet: bool,
verbose: bool,
clear_output: bool,
src: Tuple[str],
config: Optional[str],
) -> None:
"""
The uncompromising code formatter, for Jupyter notebooks.
"""
write_back = black.WriteBack.from_configuration(check=check, diff=False)
mode = black.FileMode(
target_versions=black.PY36_VERSIONS,
line_length=line_length,
is_pyi=False,
string_normalization=True,
)
if config and verbose:
black.out(f"Using configuration from {config}.", bold=False, fg="blue")
try:
include_regex = black.re_compile_maybe_verbose(include)
except re.error:
black.err(f"Invalid regular expression for include given: {include!r}")
ctx.exit(2)
try:
if changed is not black.Changed.CACHED:
sub_report = format_file_in_place(
src,
line_length=line_length,
write_back=write_back,
mode=mode,
clear_output=clear_output,
sub_report=sub_report,
)
if sub_report.change_count:
changed = black.Changed.YES
if (
write_back is black.WriteBack.YES
and changed is not black.Changed.CACHED
) or (
write_back is black.WriteBack.CHECK and changed is black.Changed.NO
):
black.write_cache(cache, [src], line_length, mode)
report.done(src, changed)
if changed is not black.Changed.CACHED:
click.secho(f" {sub_report}", err=True)
except Exception as exc:
report.failed(src, str(exc))
class_fields = [
["Field", "Type", "Description"],
["Parameters", "Type", "Description"],
]
func_fields = [
["Parameters", "Type", "Required", "Description"],
["Parameter", "Type", "Required", "Description"],
]
use_back = False
use_yapf = False
black_settings = dict(
write_back=black.WriteBack.from_configuration(check=False, diff=False),
report=black.Report(check=False, quiet=False, verbose=False),
mode=black.FileMode(
target_versions=set(),
line_length=black.DEFAULT_LINE_LENGTH,
is_pyi=False,
string_normalization=True,
),
)
yapf_settings = dict(
style={
'ALIGN_CLOSING_BRACKET_WITH_VISUAL_INDENT': True,
'ALLOW_MULTILINE_LAMBDAS': True,
'ALLOW_MULTILINE_DICTIONARY_KEYS': False,
'ALLOW_SPLIT_BEFORE_DEFAULT_OR_NAMED_ASSIGNS': True,
'ALLOW_SPLIT_BEFORE_DICT_VALUE': False,
'ARITHMETIC_PRECEDENCE_INDICATION': False,
try:
cell["outputs"], cell[
"execution_count"
] = clear_cell_outputs(
cell["outputs"], cell["execution_count"]
)
sub_report.done_output(black.Changed.YES)
except black.NothingChanged:
sub_report.done_output(black.Changed.NO)
dst_cells.append(cell)
src_contents["cells"] = dst_cells
# click.echo(src)
# click.secho(f" {sub_report}", err=True)
if write_back is black.WriteBack.YES:
with src.open("w") as fp:
nbformat.write(src_contents, fp)
return sub_report
changed = black.Changed.CACHED
if changed is not black.Changed.CACHED:
sub_report = format_file_in_place(
src,
write_back=write_back,
mode=mode,
clear_output=clear_output,
sub_report=sub_report,
)
if sub_report.change_count or sub_report.output_change_count:
changed = black.Changed.YES
if (
write_back is black.WriteBack.YES
and changed is not black.Changed.CACHED
) or (
write_back is black.WriteBack.CHECK and changed is black.Changed.NO
):
black.write_cache(cache, [src], mode)
report.done(src, changed)
if changed is not black.Changed.CACHED and (verbose or not quiet):
click.secho(f" {sub_report}", err=True)
except Exception as exc:
report.failed(src, str(exc))
def format_stdin_to_stdout(
fast: bool, *, write_back: WriteBack = WriteBack.NO, mode: FileMode
) -> bool:
"""Format file on stdin. Return True if changed.
If `write_back` is YES, write reformatted code back to stdout. If it is DIFF,
write a diff to stdout. The `mode` argument is passed to
:func:`format_file_contents`.
"""
then = datetime.utcnow()
src, encoding, newline = decode_bytes(sys.stdin.buffer.read())
dst = src
try:
dst = format_file_contents(src, fast=fast, mode=mode)
return True
except NothingChanged:
return False
def reformat_code(filename):
SortImports(filename)
src = Path(filename)
report = black.Report()
black.reformat_one(
src=src,
line_length=88,
fast=False,
write_back=black.WriteBack.YES,
mode=black.FileMode.AUTO_DETECT,
report=report,
)
def _format_with_black(output_file_name):
black.format_file_in_place(
Path(output_file_name),
mode=black.FileMode(line_length=110),
fast=False,
write_back=black.WriteBack.YES,
)
src: Path,
write_back: black.WriteBack,
mode: black.FileMode,
clear_output: bool,
report: black.Report,
quiet: bool,
verbose: bool,
) -> None:
"""Reformat a single file under `src`."""
try:
sub_report = SubReport(write_back=write_back)
changed = black.Changed.NO
cache: black.Cache = {}
if write_back is not black.WriteBack.DIFF:
cache = black.read_cache(mode)
res_src = src.resolve()
if res_src in cache and cache[res_src] == black.get_cache_info(
res_src
):
changed = black.Changed.CACHED
if changed is not black.Changed.CACHED:
sub_report = format_file_in_place(
src,
write_back=write_back,
mode=mode,
clear_output=clear_output,
sub_report=sub_report,
)
if sub_report.change_count or sub_report.output_change_count:
changed = black.Changed.YES