第10章 Serverless架构¶

import json
import boto3                   # AWS SDK，用于访问AWS服务
from datetime import datetime

def lambda_handler(event, context):
    """Lambda函数处理程序
    event: API Gateway传入的事件对象，包含HTTP请求信息
    context: Lambda运行时上下文（函数名、内存限制、剩余时间等）
    """

    # 从API Gateway v2的事件格式中提取HTTP方法和路径
    http_method = event['requestContext']['http']['method']
    path = event['requestContext']['http']['path']

    # 根据HTTP方法和路径分发请求（简易路由）
    if http_method == 'GET' and path == '/users':
        return get_users()
    elif http_method == 'POST' and path == '/users':
        return create_user(event)
    else:
        return {
            'statusCode': 404,
            'body': json.dumps({'error': 'Not found'})  # json.dumps将Python对象转为JSON字符串
        }

def get_users():
    """获取用户列表"""
    # 初始化DynamoDB资源（建议在函数外初始化以复用连接，减少冷启动开销）
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('Users')  # 指定DynamoDB表名

    # scan扫描全表（数据量大时应使用query+分页）
    response = table.scan()
    users = response.get('Items', [])

    # 返回符合API Gateway格式的响应
    return {
        'statusCode': 200,
        'body': json.dumps(users, default=str)  # default=str处理Decimal等类型
    }

def create_user(event):
    """创建用户"""
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('Users')

    # 解析请求体中的JSON数据
    body = json.loads(event['body'])  # json.loads将JSON字符串转为Python对象
    user = {
        'userId': body['userId'],      # 用户唯一标识（分区键）
        'name': body['name'],
        'email': body['email'],
        'createdAt': datetime.now().isoformat()  # ISO格式时间戳
    }

    # 写入DynamoDB（put_item为覆盖写入）
    table.put_item(Item=user)

    return {
        'statusCode': 201,             # 201 Created
        'body': json.dumps(user, default=str)
    }

import functions_framework       # Google Cloud Functions框架
from markupsafe import escape    # HTML转义，防止XSS攻击
import json

@functions_framework.http        # HTTP触发器装饰器
def hello_http(request):
    """HTTP Cloud Function
    request: Flask的Request对象，包含HTTP请求的所有信息
    """

    # 尝试从JSON请求体获取参数
    request_json = request.get_json(silent=True)  # silent=True避免解析失败抛异常
    request_args = request.args   # URL查询参数

    # 参数优先级：JSON请求体 > URL查询参数 > 默认值
    if request_json and 'name' in request_json:
        name = request_json['name']
    elif request_args and 'name' in request_args:
        name = request_args['name']
    else:
        name = 'World'

    return f'Hello {escape(name)}!'  # escape防止XSS注入

@functions_framework.cloud_event  # CloudEvent触发器装饰器（事件驱动）
def hello_gcs(event):
    """Cloud Storage Cloud Function
    当GCS存储桶中有文件上传时自动触发
    """

    data = event.data              # CloudEvent事件数据
    bucket_name = data['bucket']   # 触发事件的存储桶名称
    file_name = data['name']       # 上传的文件名

    print(f'File {file_name} uploaded to {bucket_name}')

    return 'OK'