POC 1

.github/workflows/pull_request.yml

@@ -0,0 +1,33 @@
+name: Pull Request
+on:
+  pull_request
+
+jobs:
+  build:
+    permissions:
+      pull-requests: write
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version-file: pyproject.toml
+
+      - uses: BrandonLWhite/pipx-install-action@v1.0.1
+
+      - run: poetry install
+      - run: poe test
+      - run: poetry build
+
+      # - uses: irongut/CodeCoverageSummary@v1.3.0
+      #   with:
+      #     filename: coverage.xml
+      #     badge: true
+      #     format: markdown
+      #     output: both
+
+      # - uses: marocchino/sticky-pull-request-comment@v2
+      #   with:
+      #     path: code-coverage-results.md

.github/workflows/release.yml

@@ -0,0 +1,28 @@
+name: Release
+on:
+  release:
+    types: [created]
+
+jobs:
+  build:
+    permissions:
+      pull-requests: write
+      id-token: write # Needed for pypi trusted publishing
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version-file: pyproject.toml
+
+      - uses: BrandonLWhite/pipx-install-action@v1.0.1
+
+      - run: poetry install
+      - run: poe test
+      - run: poetry version ${{ github.ref_name }}
+      - run: poetry build
+
+      - name: Publish distribution 📦 to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1

package_python_function/main.py

@@ -0,0 +1,21 @@
+import argparse
+from pathlib import Path
+
+from .packager import Packager
+
+
+def main() -> None:
+    args = parse_args()
+    project_path = Path(args.project).resolve()
+    venv_path = Path(args.venv).resolve()
+    output_path = Path(args.output).resolve()
+    packager = Packager(venv_path, project_path, output_path)
+    packager.package()
+
+
+def parse_args() -> argparse.Namespace:
+    arg_parser = argparse.ArgumentParser()
+    arg_parser.add_argument("venv", type=str)
+    arg_parser.add_argument("--project", type=str, default='pyproject.toml')
+    arg_parser.add_argument("--output", type=str, default='.')
+    return arg_parser.parse_args()

package_python_function/nested_zip_loader.py

@@ -0,0 +1,44 @@
+# AWS imposes a 10 second limit on the INIT sequence of a Lambda function.  If this time limit is reached, the process
+# is terminated and the INIT is performed again as part of the function's billable invocation.
+# Reference: https://docs.aws.amazon.com/lambda/latest/dg/lambda-runtime-environment.html
+#
+# For this reason, we can be left with an incomplete extraction and so care is taken to avoid inadverently using it.
+#
+# From https://docs.python.org/3/reference/import.html
+# "The module will exist in sys.modules before the loader executes the module code. This is crucial because the module
+# code may (directly or indirectly) import itself"
+
+# TODO: Inspired by serverless-python-requirements.
+
+def load_nested_zip() -> None:
+    from pathlib import Path
+    import sys
+    import tempfile
+    import importlib
+
+    temp_path = Path(tempfile.gettempdir())
+
+    target_package_path = temp_path / "package-python-function"
+
+    if not target_package_path.exists():
+        import zipfile
+        import shutil
+        import os
+
+        staging_package_path = temp_path / ".stage.package-python-function"
+
+        # TODO BW: Work this out.
+        if staging_package_path.exists():
+            shutil.rmtree(str(staging_package_path))
+
+        nested_zip_path = Path(__file__).parent / '.requirements.zip'
+
+        zipfile.ZipFile(str(nested_zip_path), 'r').extractall(str(staging_package_path))
+        os.rename(str(staging_package_path), str(target_package_path))  # Atomic -- TODO BW DOCME
+
+    # TODO BW: Update this comment
+    # We want our path to look like [working_dir, serverless_requirements, ...]
+    sys.path.insert(1, target_package_path)
+    importlib.reload(sys.modules[__name__])
+
+load_nested_zip()

package_python_function/packager.py

@@ -0,0 +1,81 @@
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+import zipfile
+import shutil
+
+from .python_project import PythonProject
+
+
+class Packager:
+    AWS_LAMBDA_MAX_UNZIP_SIZE = 262144000
+
+    def __init__(self, venv_path: Path, project_path: Path, output_path: Path):
+        self.project = PythonProject(project_path)
+        self.venv_path = venv_path
+        self.output_path = output_path
+        self._uncompressed_bytes = 0
+
+    @property
+    def output_file_path(self) -> Path:
+        if self.output_path.is_dir():
+            return self.output_path / f'{self.project.name}.zip'
+        return self.output_path
+
+    @property
+    def input_path(self) -> Path:
+        python_paths = list((self.venv_path / 'lib').glob('python*'))
+        if not python_paths:
+            raise Exception("input_path")
+        return python_paths[0] / 'site-packages'
+
+    def package(self) -> None:
+        print("Packaging:", self.project.path)
+        print("Output:", self.output_file_path)
+        print("Input:", self.input_path)
+        print("Entrypoint Package name:", self.project.entrypoint_package_name)
+
+        with NamedTemporaryFile() as dependencies_zip:
+            self.zip_all_dependencies(Path(dependencies_zip.name))
+
+    def zip_all_dependencies(self, target_path: Path) -> None:
+        print(f"Zipping to {target_path} ...")
+
+        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:
+                        self._uncompressed_bytes += item.stat().st_size
+                        zip_file.write(item, item.relative_to(self.input_path))
+
+            zip_dir(self.input_path)
+
+        compressed_bytes = target_path.stat().st_size
+
+        print(f"Uncompressed size: {self._uncompressed_bytes:,} bytes")
+        print(f"Compressed size: {compressed_bytes:,} bytes")
+
+        if self._uncompressed_bytes > self.AWS_LAMBDA_MAX_UNZIP_SIZE:
+            print(f"The uncompressed size of the ZIP file is greater than the AWS Lambda limit of {self.AWS_LAMBDA_MAX_UNZIP_SIZE:,} bytes.")
+            if(compressed_bytes < self.AWS_LAMBDA_MAX_UNZIP_SIZE):
+                print(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.")
+        else:
+            shutil.copy(str(target_path), str(self.output_file_path))
+
+    def generate_nested_zip(self, inner_zip_path: Path) -> None:
+        with zipfile.ZipFile(self.output_file_path, 'w') as outer_zip_file:
+            entrypoint_dir = Path(self.project.entrypoint_package_name)
+            outer_zip_file.write(
+                inner_zip_path,
+                arcname=str(entrypoint_dir / ".dependencies.zip"),
+                compresslevel=zipfile.ZIP_STORED
+            )
+            outer_zip_file.writestr(
+                str(entrypoint_dir / "__init__.py"),
+                Path(__file__).parent.joinpath("nested_zip_loader.py").read_text(),
+                compresslevel=zipfile.ZIP_DEFLATED
+            )

package_python_function/python_project.py

@@ -0,0 +1,41 @@
+from functools import cached_property
+from pathlib import Path
+from typing import Optional
+import tomllib
+
+
+class PythonProject:
+    def __init__(self, path: Path):
+        self.path = path
+        self.toml = tomllib.loads(path.read_text())
+
+    @cached_property
+    def name(self) -> str:
+        return self.find_value((
+            ('project', 'name'),
+            ('tool', 'poetry', 'name'),
+        ))
+
+    @cached_property
+    def entrypoint_package_name(self) -> str:
+        """
+        The subdirectory name in the source virtual environment's site-packages that contains the function's entrypoint
+        code.
+        """
+        # TODO : Parse out the project's package dir(s).  Use the first one if there are multiple.
+        return self.name.replace('-', '_')
+
+    def find_value(self, paths: tuple[tuple[str]]) -> str:
+        for path in paths:
+            value = self.get_value(path)
+            if value is not None:
+                return value
+        raise Exception("TODO Exception find_value")
+
+    def get_value(self, path: tuple[str]) -> Optional[str]:
+        node = self.toml
+        for name in path:
+            node = node.get(name)
+            if node is None:
+                return None
+        return node