Merge pull request #237 from Create-Python-App/feat/issue-230-copy-optimization

feat(core): optimize file copy with reflink/hardlink fallback

Author: ulises-jeremias

packages/create-python-app-core/src/create_python_app_core/loaders.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import os
 import shutil
+import subprocess
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 from jinja2 import Environment, StrictUndefined, TemplateError
 
@@ -18,6 +20,54 @@
     autoescape=False,
 )
 
+CopyMethod = Literal["reflink", "hardlink", "copy"]
+
+
+def copy_file_efficient(
+    src: Path,
+    dest: Path,
+    *,
+    allow_hardlink: bool = True,
+) -> CopyMethod:
+    """Copy a file using reflink → hardlink → ``shutil.copy2`` (CNA parity).
+
+    Hardlinks are skipped when ``allow_hardlink`` is False (e.g. files that will
+    be mutated by template rendering) or when ``CPA_COPY_HARDLINK=0``.
+    """
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    if dest.exists():
+        dest.unlink()
+
+    hardlink_ok = allow_hardlink and os.environ.get("CPA_COPY_HARDLINK", "1") != "0"
+
+    if os.name != "nt":
+        # 1) Reflink / clonefile (near-instant CoW on Btrfs/XFS/ZFS/APFS).
+        for args in (
+            ["cp", "-c", "--", str(src), str(dest)],  # macOS clonefile
+            ["cp", "--reflink=auto", "--", str(src), str(dest)],  # GNU
+        ):
+            try:
+                subprocess.run(args, check=True, capture_output=True)
+                if dest.is_file():
+                    shutil.copystat(src, dest, follow_symlinks=True)
+                    return "reflink"
+            except (FileNotFoundError, subprocess.CalledProcessError, OSError):
+                if dest.exists():
+                    dest.unlink(missing_ok=True)
+
+        # 2) Hardlink (same inode — avoided for rendered/mutated files).
+        if hardlink_ok:
+            try:
+                os.link(src, dest)
+                return "hardlink"
+            except OSError:
+                if dest.exists():
+                    dest.unlink(missing_ok=True)
+
+    # 3) Full recursive metadata-preserving copy.
+    shutil.copy2(src, dest)
+    return "copy"
+
 
 def _mode_from_path(rel: Path) -> str:
     name = rel.name
@@ -104,8 +154,9 @@ def process_file(
 
     if target.exists() and not overwrite:
         return None
-    target.parent.mkdir(parents=True, exist_ok=True)
-    shutil.copy2(src, target)
+    # Plain files: reflink → hardlink → copy2. Never hardlink .template
+    # paths (handled above); allow_hardlink stays True for immutable copies.
+    copy_file_efficient(src, target, allow_hardlink=True)
     return target
 
 

packages/create-python-app-core/tests/test_loaders.py

@@ -70,3 +70,44 @@ def test_process_file_copy(tmp_path: Path) -> None:
     written = process_file(src, dest, Path("a.txt"), context={})
     assert written == dest / "a.txt"
     assert (dest / "a.txt").read_text() == "x"
+
+
+def test_copy_file_efficient_roundtrip(tmp_path: Path) -> None:
+    from create_python_app_core.loaders import copy_file_efficient
+
+    src = tmp_path / "src.bin"
+    src.write_bytes(b"hello-copy")
+    dest = tmp_path / "nested" / "dest.bin"
+    method = copy_file_efficient(src, dest)
+    assert method in {"reflink", "hardlink", "copy"}
+    assert dest.read_bytes() == b"hello-copy"
+
+
+def test_copy_file_efficient_skips_hardlink_when_disabled(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from create_python_app_core.loaders import copy_file_efficient
+
+    monkeypatch.setenv("CPA_COPY_HARDLINK", "0")
+    src = tmp_path / "src.txt"
+    src.write_text("data")
+    dest = tmp_path / "dest.txt"
+    method = copy_file_efficient(src, dest, allow_hardlink=True)
+    assert method in {"reflink", "copy"}
+    assert dest.read_text() == "data"
+
+
+def test_copy_file_efficient_disallow_hardlink(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from create_python_app_core.loaders import copy_file_efficient
+
+    # Force past reflink by stubbing subprocess failures if needed;
+    # allow_hardlink=False must never return hardlink.
+    monkeypatch.setenv("CPA_COPY_HARDLINK", "1")
+    src = tmp_path / "src.txt"
+    src.write_text("data")
+    dest = tmp_path / "dest.txt"
+    method = copy_file_efficient(src, dest, allow_hardlink=False)
+    assert method != "hardlink"
+    assert dest.read_text() == "data"