Merge pull request psycopg#744 from dlax/async-to-sync-jobs

feat(tools): possibly run async_to_sync.py concurrently

Author: dvarrazzo

tools/async_to_sync.py

@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import os
+from concurrent.futures import ProcessPoolExecutor
 import sys
 import logging
 import subprocess as sp
@@ -59,14 +60,15 @@
 SCRIPT_NAME = os.path.basename(sys.argv[0])
 
 logger = logging.getLogger()
-logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
 
 
 def main() -> int:
     opt = parse_cmdline()
     if opt.container:
         return run_in_container(opt.container)
 
+    logging.basicConfig(level=opt.log_level, format="%(levelname)s %(message)s")
+
     current_ver = ".".join(map(str, sys.version_info[:2]))
     if current_ver != PYVER:
         logger.warning(
@@ -83,27 +85,50 @@ def main() -> int:
             PYVER,
         )
 
-    outputs = []
-    for fpin in opt.inputs:
-        fpout = fpin.parent / fpin.name.replace("_async", "")
-        outputs.append(str(fpout))
-        logger.info("converting %s", fpin)
-        with fpin.open() as f:
-            source = f.read()
+    if not opt.convert_all:
+        inputs, outputs = [], []
+        for fpin in opt.inputs:
+            fpout = fpin.parent / fpin.name.replace("_async", "")
+            if fpout.stat().st_mtime >= fpin.stat().st_mtime:
+                logger.debug("not converting %s as %s is up to date", fpin, fpout)
+                continue
+            inputs.append(fpin)
+            outputs.append(fpout)
+        if not outputs:
+            logger.warning("all output files are up to date, nothing to do")
+            return 0
+
+    else:
+        inputs = opt.inputs
+        outputs = [fpin.parent / fpin.name.replace("_async", "") for fpin in inputs]
+
+    if opt.jobs == 1:
+        logger.debug("multi-processing disabled")
+        for fpin, fpout in zip(inputs, outputs):
+            convert(fpin, fpout)
+    else:
+        with ProcessPoolExecutor(max_workers=opt.jobs) as executor:
+            executor.map(convert, inputs, outputs)
 
-        tree = ast.parse(source, filename=str(fpin))
-        tree = async_to_sync(tree, filepath=fpin)
-        output = tree_to_str(tree, fpin)
+    if opt.check:
+        return check([str(o) for o in outputs])
 
-        with fpout.open("w") as f:
-            print(output, file=f)
+    return 0
 
-        sp.check_call(["black", "-q", str(fpout)])
 
-    if opt.check:
-        return check(outputs)
+def convert(fpin: Path, fpout: Path) -> None:
+    logger.info("converting %s", fpin)
+    with fpin.open() as f:
+        source = f.read()
 
-    return 0
+    tree = ast.parse(source, filename=str(fpin))
+    tree = async_to_sync(tree, filepath=fpin)
+    output = tree_to_str(tree, fpin)
+
+    with fpout.open("w") as f:
+        print(output, file=f)
+
+    sp.check_call(["black", "-q", str(fpout)])
 
 
 def check(outputs: list[str]) -> int:
@@ -567,6 +592,29 @@ def parse_cmdline() -> Namespace:
     parser.add_argument(
         "--all", action="store_true", help="process all the files of the project"
     )
+    parser.add_argument(
+        "-B",
+        "--convert-all",
+        action="store_true",
+        help="process specified files without checking last modification times",
+    )
+    parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        metavar="N",
+        help=(
+            "process files concurrently using at most N workers; "
+            "if unspecified, the number of processors on the machine will be used"
+        ),
+    )
+    parser.add_argument(
+        "-L",
+        "--log-level",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+        default="INFO",
+        help="Logger level.",
+    )
     container = parser.add_mutually_exclusive_group()
     container.add_argument(
         "--docker",