[paved path] add lintrunner following pytorch

Author: hudeven

torchrecipes/paved_path/.lintrunner.toml

@@ -0,0 +1,21 @@
+# Black + usort
+[[linter]]
+code = 'UFMT'
+include_patterns = [
+    '**/*.py',
+]
+command = [
+    'python3',
+    'tools/linter/ufmt_linter.py',
+    '--',
+    '@{{PATHSFILE}}'
+]
+init_command = [
+    'python3',
+    'tools/linter/pip_init.py',
+    '--dry-run={{DRYRUN}}',
+    'black==22.3.0',
+    'ufmt==1.3.3',
+    'usort==1.0.2',
+]
+is_formatter = true

torchrecipes/paved_path/tools/linter/pip_init.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+import argparse
+import logging
+import os
+import subprocess
+import sys
+import time
+
+from typing import List
+
+
+def run_command(args: List[str]) -> "subprocess.CompletedProcess[bytes]":
+    logging.debug("$ %s", " ".join(args))
+    start_time = time.monotonic()
+    try:
+        return subprocess.run(args, check=True)
+    finally:
+        end_time = time.monotonic()
+        logging.debug("took %dms", (end_time - start_time) * 1000)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="pip initializer")
+    parser.add_argument(
+        "packages",
+        nargs="+",
+        help="pip packages to install",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="verbose logging",
+    )
+    parser.add_argument(
+        "--dry-run", help="do not install anything, just print what would be done."
+    )
+
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        format="<%(threadName)s:%(levelname)s> %(message)s",
+        level=logging.NOTSET if args.verbose else logging.DEBUG,
+        stream=sys.stderr,
+    )
+
+    for package in args.packages:
+        package_name, _, version = package.partition("=")
+        if version == "":
+            raise RuntimeError(
+                "Package {package_name} did not have a version specified. "
+                "Please specify a version to product a consistent linting experience."
+            )
+    pip_args = ["pip3", "install"]
+
+    # If we are in a global install, use `--user` to install so that you do not
+    # need root access in order to initialize linters.
+    #
+    # However, `pip install --user` interacts poorly with virtualenvs (see:
+    # https://bit.ly/3vD4kvl) and conda (see: https://bit.ly/3KG7ZfU). So in
+    # these cases perform a regular installation.
+    in_conda = os.environ.get("CONDA_PREFIX") is not None
+    in_virtualenv = os.environ.get("VIRTUAL_ENV") is not None
+    if not in_conda and not in_virtualenv:
+        pip_args.append("--user")
+
+    pip_args.extend(args.packages)
+
+    dry_run = args.dry_run == "1"
+    if dry_run:
+        print(f"Would have run: {pip_args}")
+        sys.exit(0)
+
+    run_command(pip_args)

torchrecipes/paved_path/tools/linter/ufmt_linter.py

@@ -0,0 +1,140 @@
+import argparse
+import concurrent.futures
+import json
+import logging
+import os
+import sys
+from enum import Enum
+from pathlib import Path
+from typing import Any, List, NamedTuple, Optional
+
+from ufmt.core import make_black_config, ufmt_string
+from usort import Config as UsortConfig
+
+IS_WINDOWS: bool = os.name == "nt"
+
+
+def eprint(*args: Any, **kwargs: Any) -> None:
+    print(*args, file=sys.stderr, flush=True, **kwargs)
+
+
+class LintSeverity(str, Enum):
+    ERROR = "error"
+    WARNING = "warning"
+    ADVICE = "advice"
+    DISABLED = "disabled"
+
+
+class LintMessage(NamedTuple):
+    path: Optional[str]
+    line: Optional[int]
+    char: Optional[int]
+    code: str
+    severity: LintSeverity
+    name: str
+    original: Optional[str]
+    replacement: Optional[str]
+    description: Optional[str]
+
+
+def as_posix(name: str) -> str:
+    return name.replace("\\", "/") if IS_WINDOWS else name
+
+
+def format_error_message(filename: str, err: Exception) -> LintMessage:
+    return LintMessage(
+        path=filename,
+        line=None,
+        char=None,
+        code="UFMT",
+        severity=LintSeverity.ADVICE,
+        name="command-failed",
+        original=None,
+        replacement=None,
+        description=(f"Failed due to {err.__class__.__name__}:\n{err}"),
+    )
+
+
+def check_file(
+    filename: str,
+) -> List[LintMessage]:
+    with open(filename, "rb") as f:
+        original = f.read().decode("utf-8")
+
+    try:
+        path = Path(filename)
+
+        usort_config = UsortConfig.find(path)
+        black_config = make_black_config(path)
+
+        # Use UFMT API to call both usort and black
+        replacement = ufmt_string(
+            path=path,
+            content=original,
+            usort_config=usort_config,
+            black_config=black_config,
+        )
+
+        if original == replacement:
+            return []
+
+        return [
+            LintMessage(
+                path=filename,
+                line=None,
+                char=None,
+                code="UFMT",
+                severity=LintSeverity.WARNING,
+                name="format",
+                original=original,
+                replacement=replacement,
+                description="Run `lintrunner -a` to apply this patch.",
+            )
+        ]
+    except Exception as err:
+        return [format_error_message(filename, err)]
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Format files with ufmt (black + usort).",
+        fromfile_prefix_chars="@",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="verbose logging",
+    )
+    parser.add_argument(
+        "filenames",
+        nargs="+",
+        help="paths to lint",
+    )
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        format="<%(threadName)s:%(levelname)s> %(message)s",
+        level=logging.NOTSET
+        if args.verbose
+        else logging.DEBUG
+        if len(args.filenames) < 1000
+        else logging.INFO,
+        stream=sys.stderr,
+    )
+
+    with concurrent.futures.ThreadPoolExecutor(
+        max_workers=os.cpu_count(),
+        thread_name_prefix="Thread",
+    ) as executor:
+        futures = {executor.submit(check_file, x): x for x in args.filenames}
+        for future in concurrent.futures.as_completed(futures):
+            try:
+                for lint_message in future.result():
+                    print(json.dumps(lint_message._asdict()), flush=True)
+            except Exception:
+                logging.critical('Failed at "%s".', futures[future])
+                raise
+
+
+if __name__ == "__main__":
+    main()