• AWS清除CloudFront缓存


    运行之前,提供AWS密钥

    import boto3
    
    
    def find_cloudfront_distribution_id(domain_name, aws_access_key_id, aws_secret_access_key):
        # 创建 CloudFront 资源
        cloudfront = boto3.client('cloudfront', aws_access_key_id=aws_access_key_id,
                                  aws_secret_access_key=aws_secret_access_key)
    
        # 获取所有 CloudFront 分配
        distributions = cloudfront.list_distributions()['DistributionList']['Items']
    
        for distribution in distributions:
            # 检查分配是否有别名
            if 'Aliases' in distribution and 'Items' in distribution['Aliases']:
                # 检查分配的域名是否包含所提供的域名
                for alias in distribution['Aliases']['Items']:
                    if alias == domain_name:
                        return distribution['Id']
    
        return None
    
    
    def invalidate_cloudfront_cache(distribution_id, paths, aws_access_key_id, aws_secret_access_key):
        # 创建 CloudFront 客户端
        cloudfront = boto3.client('cloudfront', aws_access_key_id=aws_access_key_id,
                                  aws_secret_access_key=aws_secret_access_key)
    
        # 创建失效批次
        invalidation = {
            'DistributionId': distribution_id,
            'InvalidationBatch': {
                'Paths': {
                    'Quantity': len(paths),
                    'Items': paths
                },
                'CallerReference': 'unique-identifier'  # 为失效批次提供唯一标识符
            }
        }
    
        # 提交失效请求
        response = cloudfront.create_invalidation(**invalidation)
    
    
       # 检查失效请求是否成功
        if response['ResponseMetadata']['HTTPStatusCode'] == 201:
            print("缓存失效请求成功。")
        else:
            print("缓存失效请求失败。")
    
    
    # 输入要清理的域名
    domain_name = input('请输入需要清理缓存的域名:\n')
    
    # AWS 访问密钥和秘密访问密钥
    aws_access_key_id = ''
    aws_secret_access_key = ''
    
    # 查找 CloudFront 分配的 ID
    distribution_id = find_cloudfront_distribution_id(domain_name, aws_access_key_id, aws_secret_access_key)
    
    if distribution_id:
        # 要失效的路径
        paths_to_invalidate = ['/*']
    
        # 失效缓存
        invalidate_cloudfront_cache(distribution_id, paths_to_invalidate, aws_access_key_id, aws_secret_access_key)
    
    else:
        print(f"未找到与域名 '{domain_name}' 相关联的 CloudFront 分配")
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
  • 相关阅读:
    pytest:开始使用
    JWT的无限可能性:它如何在身份验证、单点登录、API安全等领域发挥作用
    Leetcode刷题详解——四数之和
    .NET Emit 入门教程:第七部分:实战项目1:将 DbDataReader 转实体
    Neo4j 基本语法
    第十二章 类和对象——封装
    Linux下实现U盘的挂载,复制文件和卸载等操作
    19-springcloud(上)
    深入了解 XXE 注射
    JVM学习-监控工具(二)
  • 原文地址:https://blog.csdn.net/weixin_42324463/article/details/138091283