Add vpc_config to awslambda.Function

This will allow to set the VPC for lambda functions

Author: adanaja

src/e3/aws/troposphere/awslambda/__init__.py

@@ -48,6 +48,7 @@ def __init__(
         environment: dict[str, str] | None = None,
         logging_config: awslambda.LoggingConfig | None = None,
         dl_config: awslambda.DeadLetterConfig | None = None,
+        vpc_config: awslambda.VPCConfig | None = None,
     ):
         """Initialize an AWS lambda function.
 
@@ -76,6 +77,10 @@ def __init__(
         :param logging_config: The function's Amazon CloudWatch Logs settings
         :param dl_config: The dead letter config that specifies the topic or queue where
             lambda sends asynchronous events when they fail processing
+        :param vpc_config: For network connectivity to AWS resources in a VPC, specify
+            a list of security groups and subnets in the VPC. When you connect a
+            function to a VPC, it can access resources and the internet only
+            through that VPC
         """
         self.name = name
         self.description = description
@@ -94,6 +99,7 @@ def __init__(
         self.environment = environment
         self.logging_config = logging_config
         self.dl_config = dl_config
+        self.vpc_config = vpc_config
 
     def cfn_policy_document(self, stack: Stack) -> PolicyDocument:
         statements = [
@@ -209,6 +215,9 @@ def lambda_resources(
         if self.dl_config is not None:
             params["DeadLetterConfig"] = self.dl_config
 
+        if self.vpc_config is not None:
+            params["VpcConfig"] = self.vpc_config
+
         result = [awslambda.Function(name_to_id(self.name), **params)]
         # If retention duration is given provide a log group.
         # If not provided the lambda creates a log group with
@@ -392,6 +401,7 @@ def __init__(
         environment: dict[str, str] | None = None,
         logging_config: awslambda.LoggingConfig | None = None,
         dl_config: awslambda.DeadLetterConfig | None = None,
+        vpc_config: awslambda.VPCConfig | None = None,
     ):
         """Initialize an AWS lambda function with a Python runtime.
 
@@ -420,6 +430,10 @@ def __init__(
         :param logging_config: The function's Amazon CloudWatch Logs settings
         :param dl_config: The dead letter config that specifies the topic or queue where
             lambda sends asynchronous events when they fail processing
+        :param vpc_config: For network connectivity to AWS resources in a VPC, specify
+            a list of security groups and subnets in the VPC. When you connect a
+            function to a VPC, it can access resources and the internet only
+            through that VPC
         """
         assert runtime.startswith("python"), "PyFunction only accept Python runtimes"
         super().__init__(
@@ -439,6 +453,7 @@ def __init__(
             environment=environment,
             logging_config=logging_config,
             dl_config=dl_config,
+            vpc_config=vpc_config,
         )
         self.code_dir = code_dir
         self.requirement_file = requirement_file

tests/tests_e3_aws/troposphere/awslambda/awslambda_test.py

@@ -16,6 +16,7 @@
     VersionWeight,
     LoggingConfig,
     DeadLetterConfig,
+    VPCConfig,
 )
 
 from e3.aws import AWSEnv
@@ -180,6 +181,52 @@
     },
 }
 
+EXPECTED_PYFUNCTION_WITH_VPC_TEMPLATE = {
+    "Mypylambda": {
+        "Properties": {
+            "Code": {
+                "S3Bucket": "cfn_bucket",
+                "S3Key": "templates/mypylambda_lambda.zip",
+            },
+            "Description": "this is a test with vpcconfig",
+            "FunctionName": "mypylambda",
+            "Handler": "app.main",
+            "Role": "somearn",
+            "Runtime": "python3.12",
+            "Timeout": 3,
+            "VpcConfig": {
+                "SecurityGroupIds": [
+                    "sg-085912345678492fb",
+                ],
+                "SubnetIds": [
+                    "subnet-071f712345678e7c8",
+                    "subnet-07fd123456788a036",
+                ],
+            },
+            "MemorySize": 128,
+            "EphemeralStorage": {"Size": 1024},
+            "ReservedConcurrentExecutions": 1,
+            "Environment": {
+                "Variables": {"env_key_1": "env_value_1", "env_key_2": "env_value2"}
+            },
+            "LoggingConfig": {
+                "ApplicationLogLevel": "INFO",
+                "LogFormat": "JSON",
+                "SystemLogLevel": "WARN",
+            },
+        },
+        "Type": "AWS::Lambda::Function",
+    },
+    "MypylambdaLogGroup": {
+        "DeletionPolicy": "Retain",
+        "Properties": {
+            "LogGroupName": "/aws/lambda/mypylambda",
+            "RetentionInDays": 7,
+        },
+        "Type": "AWS::Logs::LogGroup",
+    },
+}
+
 EXPECTED_PYFUNCTION_POLICY_DOCUMENT = {
     "Statement": [
         {
@@ -494,6 +541,38 @@ def test_pyfunction_with_dlconfig(stack: Stack) -> None:
     assert stack.export()["Resources"] == EXPECTED_PYFUNCTION_WITH_DLQ_TEMPLATE
 
 
+def test_pyfunction_with_vpcconfig(stack: Stack) -> None:
+    stack.s3_bucket = "cfn_bucket"
+    stack.s3_key = "templates/"
+    stack.add(
+        PyFunction(
+            name="mypylambda",
+            description="this is a test with vpcconfig",
+            role="somearn",
+            runtime="python3.12",
+            code_dir="my_code_dir",
+            handler="app.main",
+            memory_size=128,
+            ephemeral_storage_size=1024,
+            logs_retention_in_days=7,
+            reserved_concurrent_executions=1,
+            environment={"env_key_1": "env_value_1", "env_key_2": "env_value2"},
+            logging_config=LoggingConfig(
+                ApplicationLogLevel="INFO",
+                LogFormat="JSON",
+                SystemLogLevel="WARN",
+            ),
+            vpc_config=VPCConfig(
+                "mypylambdavpc",
+                SecurityGroupIds=["sg-085912345678492fb"],
+                SubnetIds=["subnet-071f712345678e7c8", "subnet-07fd123456788a036"],
+            ),
+        )
+    )
+    print(stack.export()["Resources"])
+    assert stack.export()["Resources"] == EXPECTED_PYFUNCTION_WITH_VPC_TEMPLATE
+
+
 @pytest.mark.parametrize(
     "python_version, platform_list",
     [