diff --git a/packages/create-python-app-core/src/create_python_app_core/loaders.py b/packages/create-python-app-core/src/create_python_app_core/loaders.py index e7d406c..df87e34 100644 --- a/packages/create-python-app-core/src/create_python_app_core/loaders.py +++ b/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 diff --git a/packages/create-python-app-core/tests/test_loaders.py b/packages/create-python-app-core/tests/test_loaders.py index e3e21f8..5cefde7 100644 --- a/packages/create-python-app-core/tests/test_loaders.py +++ b/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"