aws-lambda-python-integration

AWS Lambda Python 集成

使用 Python 创建高性能 AWS Lambda 函数的模式，优化冷启动并采用清晰的架构。

概述

AWS Lambda Python 集成提供两种方式：AWS Chalice（功能全面的框架）和 原生 Python（开销极小）。两者均支持与 API Gateway/ALB 集成，并提供适用于生产环境的配置。

适用场景

在以下情况下使用此技能：

• 使用 Python 创建新的 Lambda 函数
• 将现有 Python 应用迁移到 Lambda
• 优化 Python Lambda 的冷启动性能
• 在基于框架的方式和精简的 Python 方式之间进行选择
• 配置 API Gateway 或 ALB 集成
• 为 Python Lambda 设置部署流水线

说明

1. 选择实现方式

实现方式	冷启动	适用场景	复杂度
AWS Chalice	< 200ms	REST API、快速开发、内置路由	低
原生 Python	< 100ms	简单处理程序、最大限度的控制、最少依赖	低

2. 项目结构

AWS Chalice 结构

my-chalice-app/
├── app.py                    # Main application with routes
├── requirements.txt          # Dependencies
├── .chalice/
│   ├── config.json          # Chalice configuration
│   └── deploy/              # Deployment artifacts
├── chalicelib/              # Additional modules
│   ├── __init__.py
│   └── services.py
└── tests/
    └── test_app.py

原生 Python 结构

my-lambda-function/
├── lambda_function.py       # Handler entry point
├── requirements.txt         # Dependencies
├── template.yaml            # SAM/CloudFormation template
└── src/                     # Additional modules
    ├── __init__.py
    ├── handlers.py
    └── utils.py

3. 实现示例

详细实现指南请参阅参考资料部分。以下是快速示例：

AWS Chalice：

from chalice import Chalice
app = Chalice(app_name='my-api')

@app.route('/')
def index():
    return {'message': 'Hello from Chalice!'}

原生 Python：

def lambda_handler(event, context):
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Hello from Lambda!'})
    }

核心概念

冷启动优化

关键策略：

1. 在模块级别初始化 - 在温启动的调用之间保持状态
2. 使用延迟加载 - 在需要时再导入开销较大的模块
3. 缓存 boto3 客户端 - 在多次调用之间复用连接

详细模式请参阅 原生 Python Lambda。

连接管理

在模块级别创建客户端并复用：

_dynamodb = None

def get_table():
    global _dynamodb
    if _dynamodb is None:
        _dynamodb = boto3.resource('dynamodb').Table('my-table')
    return _dynamodb

环境配置

class Config:
    TABLE_NAME = os.environ.get('TABLE_NAME')
    DEBUG = os.environ.get('DEBUG', 'false').lower() == 'true'

    @classmethod
    def validate(cls):
        if not cls.TABLE_NAME:
            raise ValueError("TABLE_NAME required")

最佳实践

内存和超时配置

• 内存：简单处理程序从 256MB 开始，复杂操作从 512MB 开始
• 超时：根据预期处理时间设置
  • 简单处理程序：3-5 秒
  • 包含数据库调用的 API：10-15 秒
  • 数据处理：30-60 秒

依赖项

保持 requirements.txt 精简：

# Core AWS SDK - always needed
boto3>=1.35.0

# Only add what you need
requests>=2.32.0  # If calling external APIs
pydantic>=2.5.0   # If using data validation

错误处理

返回正确的 HTTP 状态码和请求 ID：

def lambda_handler(event, context):
    try:
        result = process_event(event)
        return {'statusCode': 200, 'body': json.dumps(result)}
    except ValueError as e:
        return {'statusCode': 400, 'body': json.dumps({'error': str(e)})}
    except Exception as e:
        print(f"Error: {str(e)}")  # Log to CloudWatch
        return {'statusCode': 500, 'body': json.dumps({'error': 'Internal error'})}

结构化错误处理模式请参阅 原生 Python Lambda。

日志记录

使用结构化日志，以便通过 CloudWatch Insights 查询：

import logging, json
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Structured log
logger.info(json.dumps({
    'eventType': 'REQUEST',
    'requestId': context.aws_request_id,
    'path': event.get('path')
}))

高级模式请参阅 原生 Python Lambda。

部署选项

快速入门

**验证检查点：**部署前务必运行 serverless print 或 sam validate，以便尽早发现配置错误。

Serverless Framework：

# serverless.yml
service: my-python-api
provider:
  name: aws
  runtime: python3.12  # or python3.11
functions:
  api:
    handler: lambda_function.lambda_handler
    events:
      - http:
          path: /{proxy+}
          method: ANY

AWS SAM：

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./
      Handler: lambda_function.lambda_handler
      Runtime: python3.12  # or python3.11
      Events:
        ApiEvent:
          Type: Api
          Properties:
            Path: /{proxy+}
            Method: ANY

AWS Chalice：

chalice new-project my-api
cd my-api
chalice local 8080  # Test locally before deploying
chalice deploy --stage dev

**验证检查点：**部署到生产环境前，使用 chalice local 或 sam local invoke 在本地进行测试。

完整的部署配置（包括 CI/CD、特定环境的设置，以及高级 SAM/Serverless 模式），请参阅 Serverless 部署。

限制与警告

Lambda 限制

• 部署包：解压后最大 250MB（压缩后 50MB）
• 内存：128MB 至 10GB
• 超时：最长 15 分钟
• 并发执行数：默认 1000（可调整）
• 环境变量：总大小为 4KB

Python 特有注意事项

• 冷启动：Python 的冷启动性能出色；避免在模块级别导入开销较大的模块
• 依赖项：保持 requirements.txt 精简；使用 Lambda Layers 共享依赖项
• 原生依赖项：必须针对 Amazon Linux 2（x86_64 或 arm64）进行编译

常见陷阱

1. 在模块级别导入大型库 - 如果并非总是需要，请延迟到函数级别导入
2. 未处理 Lambda 上下文 - 使用 context.get_remaining_time_in_millis() 了解超时情况
3. 未验证输入 - 始终验证并清理事件数据
4. 打印敏感数据 - 注意日志和 CloudWatch 中的内容

错误恢复： 如果部署失败，请检查 CloudWatch 日志中的初始化错误，并运行 sam logs 诊断问题。

安全注意事项

• 切勿将凭证硬编码；使用 IAM 角色和环境变量
• 验证所有输入数据
• 使用权限最小化的 IAM 策略
• 启用 CloudTrail 进行审计日志记录

参考资料

有关特定主题的详细指南：

• AWS Chalice - 完整的 Chalice 配置、路由、中间件和部署指南
• Raw Python Lambda - 精简的处理程序模式、模块缓存和打包
• Serverless Deployment - Serverless Framework、SAM 和 CI/CD 管道
• Testing Lambda - pytest、moto、SAM Local 和 localstack

示例

示例 1：创建 AWS Chalice REST API

输入：

Create a Python Lambda REST API using AWS Chalice for a todo application

过程：

1. 使用 chalice new-project 初始化 Chalice 项目
2. 配置 CRUD 操作的路由
3. 设置 DynamoDB 集成
4. 配置部署阶段
5. 使用 chalice deploy 部署

输出：

• 完整的 Chalice 项目结构
• 包含 CRUD 端点的 REST API
• DynamoDB 表配置
• 部署配置

示例 2：优化原生 Python 的冷启动

输入：

My Python Lambda has slow cold start, how do I optimize it?

过程：

1. 分析导入语句和初始化代码
2. 将大型库的导入移至函数内部（延迟加载）
3. 在模块级别缓存 boto3 客户端
4. 移除不必要的依赖项
5. 根据需要使用预置并发

输出：

• 使用延迟加载重构后的代码
• 优化后的冷启动时间 < 100ms
• 依赖项分析

示例 3：使用 GitHub Actions 部署

输入：

Configure CI/CD for Python Lambda with SAM

过程：

1. 创建 GitHub Actions 工作流
2. 设置 Python 环境和依赖项
3. 运行 pytest 并生成覆盖率报告
4. 使用 SAM 打包
5. 部署到开发/生产阶段

输出：

• 完整的 .github/workflows/deploy.yml
• 多阶段管道
• 集成的测试自动化

版本

版本：1.0.0