import asyncio
+import logging
from concurrent.futures import Executor, ProcessPoolExecutor
from datetime import datetime
from functools import partial
-import logging
from multiprocessing import freeze_support
from typing import Set, Tuple
-from aiohttp import web
-import aiohttp_cors
+try:
+ from aiohttp import web
+ import aiohttp_cors
+except ImportError as ie:
+ raise ImportError(
+ f"aiohttp dependency is not installed: {ie}. "
+ + "Please re-install black with the '[d]' extra install "
+ + "to obtain aiohttp_cors: `pip install black[d]`"
+ ) from None
+
import black
+from black.concurrency import maybe_install_uvloop
import click
from _black_version import version as __version__
LINE_LENGTH_HEADER = "X-Line-Length"
PYTHON_VARIANT_HEADER = "X-Python-Variant"
SKIP_STRING_NORMALIZATION_HEADER = "X-Skip-String-Normalization"
+SKIP_MAGIC_TRAILING_COMMA = "X-Skip-Magic-Trailing-Comma"
FAST_OR_SAFE_HEADER = "X-Fast-Or-Safe"
DIFF_HEADER = "X-Diff"
LINE_LENGTH_HEADER,
PYTHON_VARIANT_HEADER,
SKIP_STRING_NORMALIZATION_HEADER,
+ SKIP_MAGIC_TRAILING_COMMA,
FAST_OR_SAFE_HEADER,
DIFF_HEADER,
]
skip_string_normalization = bool(
request.headers.get(SKIP_STRING_NORMALIZATION_HEADER, False)
)
+ skip_magic_trailing_comma = bool(
+ request.headers.get(SKIP_MAGIC_TRAILING_COMMA, False)
+ )
fast = False
if request.headers.get(FAST_OR_SAFE_HEADER, "safe") == "fast":
fast = True
is_pyi=pyi,
line_length=line_length,
string_normalization=not skip_string_normalization,
+ magic_trailing_comma=not skip_magic_trailing_comma,
)
req_bytes = await request.content.read()
charset = request.charset if request.charset is not None else "utf8"
def patched_main() -> None:
+ maybe_install_uvloop()
freeze_support()
black.patch_click()
main()