Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -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

Expand All@@ -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
Expand DownExpand Up@@ -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


Expand Down
41 changes: 41 additions & 0 deletions packages/create-python-app-core/tests/test_loaders.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -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"
Loading