如何使用AWS Python CDK实现CloudFront分发失效自动化？

你可以通过自定义资源（CustomResource）+ Lambda函数来实现CloudFront的失效请求，因为CDK没有直接提供创建失效的原生构造，需要借助Lambda调用AWS SDK完成操作。

步骤1：编写触发失效的Lambda函数 #

创建一个Python Lambda函数，使用boto3调用CloudFront的create_invalidation接口：

# lambda/cloudfront_invalidate.py
import boto3
import os

def handler(event, context):
    cloudfront = boto3.client('cloudfront')
    distribution_id = os.environ['DISTRIBUTION_ID']
    
    # 失效所有文件，也可以指定路径比如['/index.html', '/css/*']
    invalidation = cloudfront.create_invalidation(
        DistributionId=distribution_id,
        InvalidationBatch={
            'Paths': {
                'Quantity': 1,
                'Items': ['/*']
            },
            'CallerReference': str(context.request_id)
        }
    )
    
    return {
        'InvalidationId': invalidation['Invalidation']['Id']
    }

步骤2：在CDK栈中集成Lambda和自定义资源 #

把这个Lambda函数加入你的CDK栈，然后通过CustomResource触发失效操作，结合你已有的分发导入代码：

from aws_cdk import (
    Stack,
    aws_lambda as lambda_,
    aws_cloudfront as cloudfront,
    custom_resources as cr,
    Duration
)
from constructs import Construct

class StaticSiteStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # 你已有的分发导入代码
        distribution = cloudfront.Distribution.from_distribution_attributes(
            self,
            id="ImportedDistribution",
            domain_name="https://hgfdh.cloudfront.net",
            distribution_id="hgfdh",
        )
        
        # 创建Lambda函数
        invalidate_lambda = lambda_.Function(
            self,
            "CloudFrontInvalidateLambda",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="cloudfront_invalidate.handler",
            code=lambda_.Code.from_asset("lambda"),
            environment={
                "DISTRIBUTION_ID": distribution.distribution_id
            },
            timeout=Duration.seconds(30)
        )
        
        # 给Lambda添加CloudFront失效权限
        invalidate_lambda.add_to_role_policy(
            cr.AwsCustomResourcePolicy.from_statements([
                cr.Statement.from_actions(
                    actions=["cloudfront:CreateInvalidation"],
                    resources=[f"arn:aws:cloudfront:::{distribution.distribution_id}"]
                )
            ])
        )
        
        # 创建自定义资源，触发失效（可以绑定S3上传事件，或者部署时触发）
        cr.AwsCustomResource(
            self,
            "CloudFrontInvalidationResource",
            on_create=cr.AwsSdkCall(
                service="Lambda",
                action="invoke",
                parameters={
                    "FunctionName": invalidate_lambda.function_name,
                    "InvocationType": "RequestResponse"
                },
                physical_resource_id=cr.PhysicalResourceId.from_response("Payload")
            ),
            # 如果需要每次部署都触发失效，添加on_update
            on_update=cr.AwsSdkCall(
                service="Lambda",
                action="invoke",
                parameters={
                    "FunctionName": invalidate_lambda.function_name,
                    "InvocationType": "RequestResponse"
                },
                physical_resource_id=cr.PhysicalResourceId.from_response("Payload")
            ),
            policy=cr.AwsCustomResourcePolicy.from_statements([
                cr.Statement.from_actions(
                    actions=["lambda:InvokeFunction"],
                    resources=[invalidate_lambda.function_arn]
                )
            ])
        )

补充说明 #

• 如果你需要在S3文件上传完成后自动触发失效，可以给S3桶添加Lambda触发器，把上面的失效逻辑绑定到S3的s3:ObjectCreated:*事件上，替换自定义资源的触发方式。
• CallerReference需要是唯一值，这里用Lambda的request_id确保每次请求唯一。
• 可以修改失效路径，比如只失效特定文件而不是全部/*，减少CloudFront的失效成本。

内容的提问来源于stack exchange，提问作者Mick