import asyncio import logging from concurrent.futures import Executor, ProcessPoolExecutor from datetime import datetime, timezone from functools import partial from multiprocessing import freeze_support from typing import Set, Tuple try: from aiohttp import web from .middlewares import 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 click import black from _black_version import version as __version__ from black.concurrency import maybe_install_uvloop # This is used internally by tests to shut down the server prematurely _stop_signal = asyncio.Event() # Request headers PROTOCOL_VERSION_HEADER = "X-Protocol-Version" LINE_LENGTH_HEADER = "X-Line-Length" PYTHON_VARIANT_HEADER = "X-Python-Variant" SKIP_SOURCE_FIRST_LINE = "X-Skip-Source-First-Line" SKIP_STRING_NORMALIZATION_HEADER = "X-Skip-String-Normalization" SKIP_MAGIC_TRAILING_COMMA = "X-Skip-Magic-Trailing-Comma" PREVIEW = "X-Preview" FAST_OR_SAFE_HEADER = "X-Fast-Or-Safe" DIFF_HEADER = "X-Diff" BLACK_HEADERS = [ PROTOCOL_VERSION_HEADER, LINE_LENGTH_HEADER, PYTHON_VARIANT_HEADER, SKIP_SOURCE_FIRST_LINE, SKIP_STRING_NORMALIZATION_HEADER, SKIP_MAGIC_TRAILING_COMMA, PREVIEW, FAST_OR_SAFE_HEADER, DIFF_HEADER, ] # Response headers BLACK_VERSION_HEADER = "X-Black-Version" class InvalidVariantHeader(Exception): pass @click.command(context_settings={"help_option_names": ["-h", "--help"]}) @click.option( "--bind-host", type=str, help="Address to bind the server to.", default="localhost", show_default=True, ) @click.option( "--bind-port", type=int, help="Port to listen on", default=45484, show_default=True ) @click.version_option(version=black.__version__) def main(bind_host: str, bind_port: int) -> None: logging.basicConfig(level=logging.INFO) app = make_app() ver = black.__version__ black.out(f"blackd version {ver} listening on {bind_host} port {bind_port}") # TODO: aiohttp had an incorrect annotation for `print` argument, # It'll be fixed once aiohttp releases that code web.run_app(app, host=bind_host, port=bind_port, handle_signals=True, print=None) # type: ignore[arg-type] def make_app() -> web.Application: app = web.Application( middlewares=[cors(allow_headers=(*BLACK_HEADERS, "Content-Type"))] ) executor = ProcessPoolExecutor() app.add_routes([web.post("/", partial(handle, executor=executor))]) return app async def handle(request: web.Request, executor: Executor) -> web.Response: headers = {BLACK_VERSION_HEADER: __version__} try: if request.headers.get(PROTOCOL_VERSION_HEADER, "1") != "1": return web.Response( status=501, text="This server only supports protocol version 1" ) try: line_length = int( request.headers.get(LINE_LENGTH_HEADER, black.DEFAULT_LINE_LENGTH) ) except ValueError: return web.Response(status=400, text="Invalid line length header value") if PYTHON_VARIANT_HEADER in request.headers: value = request.headers[PYTHON_VARIANT_HEADER] try: pyi, versions = parse_python_variant_header(value) except InvalidVariantHeader as e: return web.Response( status=400, text=f"Invalid value for {PYTHON_VARIANT_HEADER}: {e.args[0]}", ) else: pyi = False versions = set() 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) ) skip_source_first_line = bool( request.headers.get(SKIP_SOURCE_FIRST_LINE, False) ) preview = bool(request.headers.get(PREVIEW, False)) fast = False if request.headers.get(FAST_OR_SAFE_HEADER, "safe") == "fast": fast = True mode = black.FileMode( target_versions=versions, is_pyi=pyi, line_length=line_length, skip_source_first_line=skip_source_first_line, string_normalization=not skip_string_normalization, magic_trailing_comma=not skip_magic_trailing_comma, preview=preview, ) req_bytes = await request.content.read() charset = request.charset if request.charset is not None else "utf8" req_str = req_bytes.decode(charset) then = datetime.now(timezone.utc) header = "" if skip_source_first_line: first_newline_position: int = req_str.find("\n") + 1 header = req_str[:first_newline_position] req_str = req_str[first_newline_position:] loop = asyncio.get_event_loop() formatted_str = await loop.run_in_executor( executor, partial(black.format_file_contents, req_str, fast=fast, mode=mode) ) # Preserve CRLF line endings nl = req_str.find("\n") if nl > 0 and req_str[nl - 1] == "\r": formatted_str = formatted_str.replace("\n", "\r\n") # If, after swapping line endings, nothing changed, then say so if formatted_str == req_str: raise black.NothingChanged # Put the source first line back req_str = header + req_str formatted_str = header + formatted_str # Only output the diff in the HTTP response only_diff = bool(request.headers.get(DIFF_HEADER, False)) if only_diff: now = datetime.now(timezone.utc) src_name = f"In\t{then}" dst_name = f"Out\t{now}" loop = asyncio.get_event_loop() formatted_str = await loop.run_in_executor( executor, partial(black.diff, req_str, formatted_str, src_name, dst_name), ) return web.Response( content_type=request.content_type, charset=charset, headers=headers, text=formatted_str, ) except black.NothingChanged: return web.Response(status=204, headers=headers) except black.InvalidInput as e: return web.Response(status=400, headers=headers, text=str(e)) except Exception as e: logging.exception("Exception during handling a request") return web.Response(status=500, headers=headers, text=str(e)) def parse_python_variant_header(value: str) -> Tuple[bool, Set[black.TargetVersion]]: if value == "pyi": return True, set() else: versions = set() for version in value.split(","): if version.startswith("py"): version = version[len("py") :] if "." in version: major_str, *rest = version.split(".") else: major_str = version[0] rest = [version[1:]] if len(version) > 1 else [] try: major = int(major_str) if major not in (2, 3): raise InvalidVariantHeader("major version must be 2 or 3") if len(rest) > 0: minor = int(rest[0]) if major == 2: raise InvalidVariantHeader("Python 2 is not supported") else: # Default to lowest supported minor version. minor = 7 if major == 2 else 3 version_str = f"PY{major}{minor}" if major == 3 and not hasattr(black.TargetVersion, version_str): raise InvalidVariantHeader(f"3.{minor} is not supported") versions.add(black.TargetVersion[version_str]) except (KeyError, ValueError): raise InvalidVariantHeader("expected e.g. '3.7', 'py3.5'") from None return False, versions def patched_main() -> None: maybe_install_uvloop() freeze_support() main() if __name__ == "__main__": patched_main()