diff --git a/src/installer/_compile_worker.py b/src/installer/_compile_worker.py new file mode 100644 index 00000000..a72e3c54 --- /dev/null +++ b/src/installer/_compile_worker.py @@ -0,0 +1,24 @@ +import compileall +from typing import Iterable, Sequence + + +def compile_files( + optimization_levels: Iterable[int], + target_and_embed_dirs: Iterable[ + Sequence[str] + ], # JSON deserialized content are all lists, so just use Sequence instead of Tuple[str, str] +) -> None: + """Perform actual compilation work.""" + for file_and_embed_dir in target_and_embed_dirs: + file, embed_dir = file_and_embed_dir + for level in optimization_levels: + # We use ``compileall`` instead of ``py_compile`` + # because ``compileall`` has heuristics to skip files which are not compilable + compileall.compile_file(file, quiet=1, ddir=embed_dir, optimize=level) + + +if __name__ == "__main__": + import json + import sys + + compile_files(**json.loads(sys.stdin.read())) diff --git a/src/installer/destinations.py b/src/installer/destinations.py index a3c1967a..94cc5159 100644 --- a/src/installer/destinations.py +++ b/src/installer/destinations.py @@ -1,8 +1,8 @@ """Handles all file writing and post-installation processing.""" -import compileall import io import os +import sys from pathlib import Path from typing import ( TYPE_CHECKING, @@ -238,18 +238,41 @@ def write_script( return entry - def _compile_bytecode(self, scheme: Scheme, record: RecordEntry) -> None: + def _get_target_and_embed_dir( + self, scheme: Scheme, record: RecordEntry + ) -> Tuple[str, str]: """Compile bytecode for a single .py file.""" - if scheme not in ("purelib", "platlib"): - return - target_path = self._path_with_destdir(scheme, record.path) dir_path_to_embed = os.path.dirname( # Without destdir os.path.join(self.scheme_dict[scheme], record.path) ) - for level in self.bytecode_optimization_levels: - compileall.compile_file( - target_path, optimize=level, quiet=1, ddir=dir_path_to_embed + return (target_path, dir_path_to_embed) + + def _perform_compilation( + self, target_and_embed_dirs: Iterable[Tuple[str, str]] + ) -> None: + """Perform actual compilation work.""" + import installer._compile_worker + + if self.interpreter == sys.executable: + installer._compile_worker.compile_files( + self.bytecode_optimization_levels, target_and_embed_dirs + ) + + else: + import inspect + import json + import subprocess + + source_code = inspect.getsource(installer._compile_worker) + input_params = { + "optimization_levels": self.bytecode_optimization_levels, + "target_and_embed_dirs": target_and_embed_dirs, + } + subprocess.run( + [self.interpreter, "-c", source_code], + check=True, + input=json.dumps(input_params, ensure_ascii=True).encode(), ) def finalize_installation( @@ -280,5 +303,9 @@ def prefix_for_scheme(file_scheme: str) -> Optional[str]: scheme, record_file_path, record_stream, is_executable=False ) - for scheme, record in record_list: - self._compile_bytecode(scheme, record) + target_and_embed_dirs = [ + self._get_target_and_embed_dir(scheme, record) + for scheme, record in record_list + if scheme in ("purelib", "platlib") + ] + self._perform_compilation(target_and_embed_dirs)