Custom Resource to make lambda functions with pip packages and more than 4096 bytes of code

codelayer.py

# Code accompanying https://blog.claude.nl/tech/howto/2021/02/15/better-lambdas-with-pip-packages-in-cloudformation.html
import json
import logging
import pathlib
import re
import subprocess as sp
import sys
import tempfile
import typing as t
import shutil

import cfnresponse as cfnr
import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)


class CodeLayerException(Exception):
    pass


def _create(properties) -> t.Tuple[str, t.Mapping[str, str]]:
    layername = properties["LayerName"]
    description = properties.get("Description", "CodeLayer")
    files = properties.get("Files", {})
    packages = properties.get("Packages", [])
    description += " ({})".format(", ".join(packages))

    assert isinstance(layername, str)
    assert isinstance(description, str)
    assert isinstance(packages, list) \
            and all(isinstance(p, str) for p in packages)
    assert isinstance(files, dict) and all(isinstance(k, str)
                                           for _k, _v in files.items()
                                           for k in (_k, _v))

    if not packages and not files:
        raise CodeLayerException("Cowardly refusing to create empty layer")

    with tempfile.TemporaryDirectory() as parenttmpdir:
        tempdir = pathlib.Path(parenttmpdir) / "python"
        if packages:
            try:
                sp.check_call([sys.executable, "-m", "pip", "install",
                                       *packages, "-t", tempdir])
            except sp.CalledProcessError:
                raise CodeLayerException("Error while installing %s" % str(packages))
        for filename, content in files.items():
            filepath = tempdir / filename
            filepath.parent.mkdir(parents=True, exist_ok=True)
            filepath.write_text(content)
            if filepath.suffix == ".py":
                try:
                    sp.check_call([
                        sys.executable, "-m", "py_compile", filepath])
                except sp.CalledProcessError:
                    raise CodeLayerException(
                        "File %s is not valid python" % filename)

        zipfilename = pathlib.Path(tempfile.NamedTemporaryFile(suffix=".zip").name)
        shutil.make_archive(
            zipfilename.with_suffix(""), format="zip", root_dir=parenttmpdir)

        client = boto3.client("lambda")
        layer = client.publish_layer_version(
            LayerName=layername,
            Description=description,
            Content={"ZipFile": zipfilename.read_bytes()},
            CompatibleRuntimes=["python%d.%d" % sys.version_info[:2]],
        )

    return (layer["LayerVersionArn"], {})

def _delete(physical_id):
    match = re.fullmatch(
        r"arn:aws:lambda:(?P<region>[^:]+):(?P<account>\d+):layer:"
        r"(?P<layername>[^:]+):(?P<version_number>\d+)", physical_id)
    if not match:
        logger.warning("Cannot parse physical id %s, not deleting", physical_id)
        return
    layername = match.group("layername")
    version_number = int(match.group("version_number"))

    client = boto3.client("lambda")
    client.delete_layer_version(
        LayerName=layername,
        VersionNumber=version_number)


def handler(e, c):
    try:
        if e["RequestType"].upper() in ("CREATE", "UPDATE"):
            physicalId, attributes = _create(e["ResourceProperties"])
            cfnr.send(
                event=e,
                context=c,
                responseData=attributes,
                responseStatus=cfnr.SUCCESS,
                physicalResourceId=physicalId,
            )
        else:
            assert e["RequestType"].upper() == "DELETE"
            _delete(e["PhysicalResourceId"])
            cfnr.send(
                event=e,
                context=c,
                responseData={},
                responseStatus=cfnr.SUCCESS,
                physicalResourceId=e["PhysicalResourceId"],
            )
    except Exception as exc:
        logger.exception("Internal Error")
        cfnr.send(
            event=e,
            context=c,
            responseData=None,
            responseStatus=cfnr.FAILED,
            reason=str(exc))

codelayer.yaml

CodeLayerLambdaRole:
  Type: AWS::IAM::Role
  Properties:
    AssumeRolePolicyDocument:
      Version: 2012-10-17
      Statement:
        - Action:
            - sts:AssumeRole
          Effect: Allow
          Principal:
            Service:
              - lambda.amazonaws.com
    Policies:
      - PolicyDocument:
          Version: 2012-10-17
          Statement:
            - Action:
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
              Effect: Allow
              Resource:
                - !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/CodeLayer-${AWS::StackName}:*
            - Action:
              - lambda:PublishLayerVersion
              - lambda:DeleteLayerVersion
              Effect: Allow
              Resource:
                - "*"
        PolicyName: lambda

CodeLayerLambda:
  Type: AWS::Lambda::Function
  Properties:
    Description: Create layers based on pip and code
    FunctionName: !Sub "CodeLayer-${AWS::StackName}"
    Handler: index.handler
    MemorySize: 1024
    Role: !GetAtt CodeLayerLambdaRole.Arn
    Runtime: python3.8
    Timeout: 300
    Code:
      ZipFile: |
        __INCLUDE__: codelayer.py