diff --git a/package_python_function/packager.py b/package_python_function/packager.py index b3d1d7d..20687d0 100644 --- a/package_python_function/packager.py +++ b/package_python_function/packager.py @@ -1,15 +1,21 @@ +from __future__ import annotations + +import logging +import os +import shutil +import time +import zipfile from pathlib import Path from tempfile import NamedTemporaryFile -import zipfile -import shutil -import logging +from typing import TYPE_CHECKING from .python_project import PythonProject +if TYPE_CHECKING: + from typing import Tuple logger = logging.getLogger(__name__) - class Packager: AWS_LAMBDA_MAX_UNZIP_SIZE = 262144000 @@ -40,14 +46,34 @@ def package(self) -> None: def zip_all_dependencies(self, target_path: Path) -> None: logger.info(f"Zipping to {target_path}...") - with zipfile.ZipFile(target_path, 'w', zipfile.ZIP_DEFLATED) as zip_file: + def date_time() -> Tuple[int, int, int, int, int, int]: + """Returns date_time value used to force overwrite on all ZipInfo objects. Defaults to + 1980-01-01 00:00:00. You can set this with the environment variable SOURCE_DATE_EPOCH as an + integer value representing seconds since Epoch. + """ + source_date_epoch = os.environ.get("SOURCE_DATE_EPOCH", None) + if source_date_epoch is not None: + return time.gmtime(int(source_date_epoch))[:6] + return (1980, 1, 1, 0, 0, 0) + + with zipfile.ZipFile(target_path, "w", zipfile.ZIP_DEFLATED) as zip_file: + def zip_dir(path: Path) -> None: for item in path.iterdir(): if item.is_dir(): zip_dir(item) else: + zinfo = zipfile.ZipInfo.from_file( + item, item.relative_to(self.input_path) + ) + zinfo.date_time = date_time() + zinfo.external_attr = 0o644 << 16 self._uncompressed_bytes += item.stat().st_size - zip_file.write(item, item.relative_to(self.input_path)) + with ( + open(item, "rb") as src, + zip_file.open(zinfo, "w") as dest, + ): + shutil.copyfileobj(src, dest, 1024 * 8) zip_dir(self.input_path) @@ -61,7 +87,7 @@ def zip_dir(path: Path) -> None: logger.info(f"The compressed size ({compressed_bytes:,}) is less than the AWS limit, so the nested-zip strategy will be used.") self.generate_nested_zip(target_path) else: - print(f"TODO Error. The unzipped size it too large for AWS Lambda.") + print("TODO Error. The unzipped size it too large for AWS Lambda.") else: logger.info(f"Copying '{target_path}' to '{self.output_file}'") shutil.copy(str(target_path), str(self.output_file)) @@ -80,4 +106,4 @@ def generate_nested_zip(self, inner_zip_path: Path) -> None: str(entrypoint_dir / "__init__.py"), Path(__file__).parent.joinpath("nested_zip_loader.py").read_text(), compresslevel=zipfile.ZIP_DEFLATED - ) \ No newline at end of file + ) diff --git a/tests/projects/project-1/pyproject.toml b/tests/projects/project-1/pyproject.toml index 1ce14d2..a5f3853 100644 --- a/tests/projects/project-1/pyproject.toml +++ b/tests/projects/project-1/pyproject.toml @@ -5,7 +5,7 @@ description = "project-1" authors = [{ name = "Brandon White", email = "brandonlwhite@gmail.com" }] license = "MIT" readme = "README.md" -requires-python = "^3.10" +requires-python = ">=3.10,<4.0" [build-system] diff --git a/tests/test_package_python_function.py b/tests/test_package_python_function.py index 5d0ba3a..32a30a5 100644 --- a/tests/test_package_python_function.py +++ b/tests/test_package_python_function.py @@ -1,12 +1,15 @@ -from pathlib import Path import sys -from package_python_function.main import main +import zipfile +from pathlib import Path +from package_python_function.main import main PROJECTS_DIR_PATH = Path(__file__).parent / 'projects' - def test_package_python_function(tmp_path: Path) -> None: + EXPECTED_FILE_MODE = 0o644 + EXPECTED_FILE_DATE_TIME = (1980, 1, 1, 0, 0, 0) + project_file_path = PROJECTS_DIR_PATH / 'project-1' / 'pyproject.toml' venv_dir_path = tmp_path / 'venv' @@ -34,4 +37,19 @@ def test_package_python_function(tmp_path: Path) -> None: ] main() - assert (output_dir_path / 'project_1.zip').exists() \ No newline at end of file + zip_file = output_dir_path / "project_1.zip" + assert zip_file.exists() + + verify_dir = tmp_path / "verify" + verify_dir.mkdir() + with zipfile.ZipFile(zip_file, "r") as zip: + zip.extractall(verify_dir) + for file_info in zip.infolist(): + mode = (file_info.external_attr >> 16) & 0xFFFF + assert mode == EXPECTED_FILE_MODE + assert file_info.date_time == EXPECTED_FILE_DATE_TIME + + assert (verify_dir / "project_1" / "__init__.py").exists() + assert (verify_dir / "project_1" / "project1.py").exists() + assert (verify_dir / "small_dependency" / "__init__.py").exists() + assert (verify_dir / "small_dependency" / "small_dependency.py").exists()