Split file utils into dedicated submodule

cibuildwheel/macos.py

@@ -31,14 +31,12 @@
     call,
     combine_constraints,
     detect_ci_provider,
-    download,
     find_compatible_wheel,
     find_uv,
     free_thread_enable_313,
     get_build_verbosity_extra_flags,
     get_pip_version,
     install_certifi_script,
-    move_file,
     prepare_command,
     read_python_configs,
     shell,
@@ -47,6 +45,7 @@
     unwrap,
     virtualenv,
 )
+from .util.files import download, move_file
 
 
 @functools.lru_cache(maxsize=None)

cibuildwheel/pyodide.py

@@ -21,19 +21,17 @@
     BuildSelector,
     call,
     combine_constraints,
-    download,
     ensure_node,
-    extract_zip,
     find_compatible_wheel,
     get_pip_version,
-    move_file,
     prepare_command,
     read_python_configs,
     shell,
     split_config_settings,
     test_fail_cwd_file,
     virtualenv,
 )
+from .util.files import download, extract_zip, move_file
 
 
 @dataclass(frozen=True)

cibuildwheel/util/__init__.py

@@ -0,0 +1,23 @@
+# flake8: noqa: F401, F403, F405
+
+
+from .files import chdir
+from .misc import *
+
+__all__ = [
+    "MANYLINUX_ARCHS",
+    "call",
+    "chdir",
+    "combine_constraints",
+    "find_compatible_wheel",
+    "find_uv",
+    "format_safe",
+    "get_build_verbosity_extra_flags",
+    "prepare_command",
+    "read_python_configs",
+    "resources_dir",
+    "selector_matches",
+    "shell",
+    "split_config_settings",
+    "strtobool",
+]

cibuildwheel/util/files.py

@@ -0,0 +1,119 @@
+"""File handling functions with default case and error handling."""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import shutil
+import ssl
+import tarfile
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+from time import sleep
+from typing import Generator
+from zipfile import ZipFile
+
+import certifi
+
+
+def extract_zip(zip_src: Path, dest: Path) -> None:
+    """Extracts a zip and correctly sets permissions on extracted files.
+
+    Note:
+      - sets permissions to the same values as they were set in the archive
+      - files with no clear permissions in `external_attr` will be extracted with default values
+    """
+    with ZipFile(zip_src) as zip_:
+        for zinfo in zip_.filelist:
+            zip_.extract(zinfo, dest)
+
+            # We have to do this manually due to https://github.com/python/cpython/issues/59999
+            permissions = (zinfo.external_attr >> 16) & 0o777
+            if permissions != 0:
+                dest.joinpath(zinfo.filename).chmod(permissions)
+
+
+def extract_tar(tar_src: Path, dest: Path) -> None:
+    """Extracts a tar file using the stdlib 'tar' filter.
+
+    See: https://docs.python.org/3/library/tarfile.html#tarfile.tar_filter for filter details
+    """
+    with tarfile.open(tar_src) as tar_:
+        tar_.extraction_filter = getattr(tarfile, "tar_filter", (lambda member, _: member))
+        tar_.extractall(dest)
+
+
+def download(url: str, dest: Path) -> None:
+    print(f"+ Download {url} to {dest}")
+    dest_dir = dest.parent
+    if not dest_dir.exists():
+        dest_dir.mkdir(parents=True)
+
+    # we've had issues when relying on the host OS' CA certificates on Windows,
+    # so we use certifi (this sounds odd but requests also does this by default)
+    cafile = os.environ.get("SSL_CERT_FILE", certifi.where())
+    context = ssl.create_default_context(cafile=cafile)
+    repeat_num = 3
+    for i in range(repeat_num):
+        try:
+            with urllib.request.urlopen(url, context=context) as response:
+                dest.write_bytes(response.read())
+                return
+
+        except OSError:
+            if i == repeat_num - 1:
+                raise
+            sleep(3)
+
+
+def move_file(src_file: Path, dst_file: Path) -> Path:
+    """Moves a file safely while avoiding potential semantic confusion:
+
+    1. `dst_file` must point to the target filename, not a directory
+    2. `dst_file` will be overwritten if it already exists
+    3. any missing parent directories will be created
+
+    Returns the fully resolved Path of the resulting file.
+
+    Raises:
+        NotADirectoryError: If any part of the intermediate path to `dst_file` is an existing file
+        IsADirectoryError: If `dst_file` points directly to an existing directory
+    """
+    src_file = src_file.resolve(strict=True)
+    dst_file = dst_file.resolve()
+
+    if dst_file.is_dir():
+        msg = "dst_file must be a valid target filename, not an existing directory."
+        raise IsADirectoryError(msg)
+    dst_file.unlink(missing_ok=True)
+    dst_file.parent.mkdir(parents=True, exist_ok=True)
+
+    # using shutil.move() as Path.rename() is not guaranteed to work across filesystem boundaries
+    # explicit str() needed for Python 3.8
+    resulting_file = shutil.move(str(src_file), str(dst_file))
+    return Path(resulting_file).resolve(strict=True)
+
+
+@dataclass(frozen=True)
+class FileReport:
+    """Caches basic details about a file to avoid repeated calls to `stat()`."""
+
+    name: str
+    size: str
+
+
+# Required until end of Python 3.10 support
+@contextlib.contextmanager
+def chdir(new_path: Path | str) -> Generator[None, None, None]:
+    """Non thread-safe context manager to temporarily change the current working directory.
+
+    Equivalent to `contextlib.chdir` in Python 3.11
+    """
+
+    cwd = os.getcwd()
+    try:
+        os.chdir(new_path)
+        yield
+    finally:
+        os.chdir(cwd)

cibuildwheel/util.py → cibuildwheel/util/misc.py

@@ -7,57 +7,34 @@
 import re
 import shlex
 import shutil
-import ssl
 import subprocess
 import sys
-import tarfile
 import textwrap
 import time
 import typing
-import urllib.request
 from collections import defaultdict
 from collections.abc import Generator, Iterable, Mapping, MutableMapping, Sequence
 from dataclasses import dataclass
 from enum import Enum
 from functools import lru_cache
 from pathlib import Path, PurePath
 from tempfile import TemporaryDirectory
-from time import sleep
 from typing import Any, ClassVar, Final, Literal, TextIO, TypeVar
-from zipfile import ZipFile
 
 import bracex
-import certifi
 from filelock import FileLock
 from packaging.requirements import InvalidRequirement, Requirement
 from packaging.specifiers import SpecifierSet
 from packaging.utils import parse_wheel_filename
 from packaging.version import Version
 from platformdirs import user_cache_path
 
-from ._compat import tomllib
-from .architecture import Architecture
-from .typing import PathOrStr, PlatformName
-
-__all__ = [
-    "MANYLINUX_ARCHS",
-    "call",
-    "chdir",
-    "combine_constraints",
-    "find_compatible_wheel",
-    "find_uv",
-    "format_safe",
-    "get_build_verbosity_extra_flags",
-    "prepare_command",
-    "read_python_configs",
-    "resources_dir",
-    "selector_matches",
-    "shell",
-    "split_config_settings",
-    "strtobool",
-]
-
-resources_dir: Final[Path] = Path(__file__).parent / "resources"
+from .._compat import tomllib
+from ..architecture import Architecture
+from ..typing import PathOrStr, PlatformName
+from .files import FileReport, download, extract_tar, extract_zip
+
+resources_dir: Final[Path] = Path(__file__).parents[1] / "resources"
 
 install_certifi_script: Final[Path] = resources_dir / "install_certifi.py"
 
@@ -321,78 +298,6 @@ def __getattr__(self, attr: str) -> Any:
         return getattr(self.stream, attr)
 
 
-def download(url: str, dest: Path) -> None:
-    print(f"+ Download {url} to {dest}")
-    dest_dir = dest.parent
-    if not dest_dir.exists():
-        dest_dir.mkdir(parents=True)
-
-    # we've had issues when relying on the host OS' CA certificates on Windows,
-    # so we use certifi (this sounds odd but requests also does this by default)
-    cafile = os.environ.get("SSL_CERT_FILE", certifi.where())
-    context = ssl.create_default_context(cafile=cafile)
-    repeat_num = 3
-    for i in range(repeat_num):
-        try:
-            with urllib.request.urlopen(url, context=context) as response:
-                dest.write_bytes(response.read())
-                return
-
-        except OSError:
-            if i == repeat_num - 1:
-                raise
-            sleep(3)
-
-
-def extract_zip(zip_src: Path, dest: Path) -> None:
-    with ZipFile(zip_src) as zip_:
-        for zinfo in zip_.filelist:
-            zip_.extract(zinfo, dest)
-
-            # Set permissions to the same values as they were set in the archive
-            # We have to do this manually due to
-            # https://github.com/python/cpython/issues/59999
-            # But some files in the zipfile seem to have external_attr with 0
-            # permissions. In that case just use the default value???
-            permissions = (zinfo.external_attr >> 16) & 0o777
-            if permissions != 0:
-                dest.joinpath(zinfo.filename).chmod(permissions)
-
-
-def extract_tar(tar_src: Path, dest: Path) -> None:
-    with tarfile.open(tar_src) as tar_:
-        tar_.extraction_filter = getattr(tarfile, "tar_filter", (lambda member, _: member))
-        tar_.extractall(dest)
-
-
-def move_file(src_file: Path, dst_file: Path) -> Path:
-    """Moves a file safely while avoiding potential semantic confusion:
-
-    1. `dst_file` must point to the target filename, not a directory
-    2. `dst_file` will be overwritten if it already exists
-    3. any missing parent directories will be created
-
-    Returns the fully resolved Path of the resulting file.
-
-    Raises:
-        NotADirectoryError: If any part of the intermediate path to `dst_file` is an existing file
-        IsADirectoryError: If `dst_file` points directly to an existing directory
-    """
-    src_file = src_file.resolve(strict=True)
-    dst_file = dst_file.resolve()
-
-    if dst_file.is_dir():
-        msg = "dst_file must be a valid target filename, not an existing directory."
-        raise IsADirectoryError(msg)
-    dst_file.unlink(missing_ok=True)
-    dst_file.parent.mkdir(parents=True, exist_ok=True)
-
-    # using shutil.move() as Path.rename() is not guaranteed to work across filesystem boundaries
-    # explicit str() needed for Python 3.8
-    resulting_file = shutil.move(str(src_file), str(dst_file))
-    return Path(resulting_file).resolve(strict=True)
-
-
 class DependencyConstraints:
     def __init__(self, base_file_path: Path):
         assert base_file_path.exists()
@@ -510,12 +415,6 @@ def unwrap(text: str) -> str:
     return re.sub(r"\s+", " ", text)
 
 
-@dataclass(frozen=True)
-class FileReport:
-    name: str
-    size: str
-
-
 @contextlib.contextmanager
 def print_new_wheels(msg: str, output_dir: Path) -> Generator[None, None, None]:
     """
@@ -772,19 +671,6 @@ def find_compatible_wheel(wheels: Sequence[T], identifier: str) -> T | None:
     return None
 
 
-# Can be replaced by contextlib.chdir in Python 3.11
-@contextlib.contextmanager
-def chdir(new_path: Path | str) -> Generator[None, None, None]:
-    """Non thread-safe context manager to change the current working directory."""
-
-    cwd = os.getcwd()
-    try:
-        os.chdir(new_path)
-        yield
-    finally:
-        os.chdir(cwd)
-
-
 def fix_ansi_codes_for_github_actions(text: str) -> str:
     """
     Github Actions forgets the current ANSI style on every new line. This

cibuildwheel/windows.py

@@ -27,13 +27,10 @@
     BuildSelector,
     call,
     combine_constraints,
-    download,
-    extract_zip,
     find_compatible_wheel,
     find_uv,
     get_build_verbosity_extra_flags,
     get_pip_version,
-    move_file,
     prepare_command,
     read_python_configs,
     shell,
@@ -42,6 +39,7 @@
     unwrap,
     virtualenv,
 )
+from .util.files import download, extract_zip, move_file
 
 
 def get_nuget_args(

unit_test/download_test.py

@@ -5,7 +5,7 @@
 import certifi
 import pytest
 
-from cibuildwheel.util import download
+from cibuildwheel.util.files import download
 
 DOWNLOAD_URL = "https://raw.githubusercontent.com/pypa/cibuildwheel/v1.6.3/requirements-dev.txt"