src/regit/__main__.py

@@ -4,9 +4,6 @@ from pathlib import Path
 from typing import Callable
 import csv
 import logging
-import os
-import shutil
-import subprocess
 import tempfile
 
 import click
@@ -14,31 +11,30 @@ import git
 import git_filter_repo as fr
 
 from .types import Executable, Repository
+from . import utils
 
 log = logging.getLogger("regit")
 
 
-def popen(cmd, **kwargs):
-    """A wrapper around subprocess.Popen that logs the command."""
-    log.debug("Running cmd %s", cmd)
-    return subprocess.Popen(cmd, **kwargs)
-
-
 @dataclass
 class BlobHandler:
+    from subprocess import Popen
+
     repo: git.Repo
     blobs_handled: dict[int, int] = field(default_factory=dict)
     is_relevant: Callable[[Path], bool] = lambda _: True
     transform: Callable[[Path, bytes], bytes] = lambda _, b: b
     filter: fr.RepoFilter | None = None
-    gitcat: subprocess.Popen | None = None
+    gitcat: Popen | None = None
 
     def __enter__(self) -> "BlobHandler":
+        from subprocess import PIPE
+
         log.debug("Starting blob handler")
-        self.gitcat = popen(
+        self.gitcat = utils.popen(
             ["git", "-C", self.repo.working_dir, "cat-file", "--batch"],
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
+            stdin=PIPE,
+            stdout=PIPE,
         )
         if self.gitcat.stdin is None or self.gitcat.stdout is None:
             raise RuntimeError("Could not start git cat-file")
@@ -78,15 +74,6 @@ class BlobHandler:
         return stdout.read(int(objsize) + 1)[:-1]
 
 
-@contextmanager
-def tfile(folder: Path, filename, content):
-    tmp_file = folder / filename
-    with open(tmp_file, "wb") as f:
-        f.write(content)
-    yield tmp_file
-    os.remove(tmp_file)
-
-
 @click.command()
 @click.option(
     "--repo",
@@ -118,17 +105,8 @@ def tfile(folder: Path, filename, content):
     help="The verbosity of logging",
     count=True,
 )
-@click.argument(
-    "program",
-    # help="The program to execute on each file",
-    type=Executable(),
-)
-@click.argument(
-    "args",
-    # help="The argurments, choose '{}' to mean the file in question, omit a '{}' to use stdin",
-    nargs=-1,
-    type=str,
-)
+@click.argument("program", type=Executable())
+@click.argument("args", nargs=-1, type=str)
 def regit(
     repo,
     pattern: str | None,
@@ -155,37 +133,37 @@ def regit(
         log.debug("Cloned repo to %s", output)
 
     with tempfile.TemporaryDirectory() as folder:
+        folder = Path(folder)
 
         def transformer(file: Path, content: bytes):
             pargs = list(args)
             try:
                 ix = pargs.index("{}")
             except ValueError:
-                cmd = [program] + pargs
-                log.debug("Running cmd %s", cmd)
-                return subprocess.run(
-                    cmd, input=content, capture_output=True, check=True
+                return utils.run(
+                    [program] + pargs, input=content, capture_output=True, check=True
                 ).stdout
             else:
-                with tfile(Path(folder), file.name, content) as tmp_file:
+                with utils.tfile(folder, file.name, content) as tmp_file:
                     pargs[ix] = str(tmp_file)
-                    cmd = [program] + pargs
-                    log.debug("Running cmd %s", cmd)
-                    subprocess.run(cmd, check=True)
+                    utils.run([program] + pargs, check=True)
                     with open(tmp_file, "rb") as f:
                         return f.read()
 
-        handler = BlobHandler(
-            repo,
-            is_relevant=lambda a: pattern is None or a.match(pattern),
-            transform=transformer,
-        )
+        def is_relevant(file: Path):
+            if pattern is None:
+                return True
+            match = file.match(pattern)
+            log.debug("Check if %s matched pattern %s", file, match)
+            return match
+
+        handler = BlobHandler(repo, is_relevant=is_relevant, transform=transformer)
         log.debug("Starting handler")
         with handler:
             options = fr.FilteringOptions.parse_args([], error_on_empty=False)
             filter = fr.RepoFilter(options, commit_callback=handler)
             handler.filter = filter
-            with chdir(repo.working_dir):
+            with utils.chdir(repo.working_dir):
                 log.debug("Starting git filter")
                 filter.run()
                 log.debug("Finished git filter")
@@ -194,29 +172,10 @@ def regit(
             writer = csv.writer(mapping)
             writer.writerow(["from", "to"])
             for fm, to in filter._commit_renames.items():
-                writer.writerow([handle(m) for m in [fm, to]])
+                writer.writerow([utils.commit_id_to_string(m) for m in [fm, to]])
 
     print(repo.working_dir)
 
 
-@contextmanager
-def chdir(path: Path):
-    old = Path.cwd()
-    log.debug("Changing dir from %s to %s", old, path)
-    os.chdir(path)
-    yield
-    log.debug("Changing dir back to %s", old)
-    os.chdir(old)
-
-
-def handle(a):
-    """Handle the result of a git renames of commits."""
-    if isinstance(a, bytes):
-        return a.decode()
-    else:
-        log.warning("Unexpected result %s", a)
-        return str(a)
-
-
 if __name__ == "__main__":
     regit()

src/regit/utils.py

+from contextlib import contextmanager
+from pathlib import Path
+import logging
+import os
+import subprocess
+
+log = logging.getLogger("regit")
+
+
+def run(cmd, **kwargs):
+    """A wrapper around subprocess.run that logs the command."""
+    log.debug("Running cmd %s", cmd)
+    return subprocess.run(cmd, **kwargs)
+
+
+def popen(cmd, **kwargs):
+    """A wrapper around subprocess.Popen that logs the command."""
+    log.debug("Running cmd %s", cmd)
+    return subprocess.Popen(cmd, **kwargs)
+
+
+@contextmanager
+def chdir(path: Path):
+    """Change the current working directory to the given path and back when done."""
+    old = Path.cwd()
+    log.debug("Changing dir from %s to %s", old, path)
+    os.chdir(path)
+    yield
+    log.debug("Changing dir back to %s", old)
+    os.chdir(old)
+
+
+def commit_id_to_string(commit_id):
+    """Convert a commit id to a string."""
+    if isinstance(commit_id, bytes):
+        return commit_id.decode()
+    else:
+        log.warning("Unexpected result %s", commit_id)
+        return str(commit_id)
+
+
+@contextmanager
+def tfile(folder: Path, filename, content):
+    """Create a temporary file with the given content and delete it when done."""
+    tmp_file = folder / filename
+    with open(tmp_file, "wb") as f:
+        f.write(content)
+    yield tmp_file
+    os.remove(tmp_file)