EdgeOne Makers Cloud Functions Python

Python 在数据处理、AI 推理、自动化脚本这些场景下用起来特别顺手。EdgeOne Makers 的 Cloud Functions 支持 Python 运行时，让你可以直接在云端跑 Python 代码。这一篇讲 Python 函数的开发流程、依赖管理和一个完整的示例。

项目结构

Python Cloud Function 的项目结构很简单。

my-function/
├── index.py          # 入口文件
├── requirements.txt  # 依赖声明
└── lib/              # 其他模块(可选)
    └── utils.py

index.py 是入口文件，requirements.txt 声明第三方依赖。

入口函数

入口函数固定叫 main_handler，接收 event 和 context 两个参数。

# index.py
def main_handler(event, context):
    """
    event: 触发事件信息，HTTP 触发时包含请求数据
    context: 运行上下文，包含函数名、请求 ID 等
    """
    print('函数被触发了')

    return {
        "statusCode": 200,
        "headers": {
            "content-type": "application/json; charset=utf-8"
        },
        "body": '{"message": "Hello Python Cloud Function"}'
    }

返回值是一个字典，平台把它转成 HTTP 响应。

处理 HTTP 请求

读取请求信息

# index.py
import json

def main_handler(event, context):
    # HTTP 触发时的关键字段
    method = event.get('httpMethod', '')
    path = event.get('path', '')
    headers = event.get('headers', {})
    query_params = event.get('queryStringParameters', {}) or {}
    body_str = event.get('body', '')

    # 读取查询参数
    keyword = query_params.get('keyword', '没有传关键词')

    # 读取请求头
    content_type = headers.get('content-type', '未知')

    return {
        "statusCode": 200,
        "headers": {"content-type": "application/json; charset=utf-8"},
        "body": json.dumps({
            "method": method,
            "path": path,
            "keyword": keyword,
            "contentType": content_type
        }, ensure_ascii=False)
    }

处理 POST 请求的 JSON 数据

# index.py
import json

def main_handler(event, context):
    if event.get('httpMethod') != 'POST':
        return {
            "statusCode": 405,
            "body": json.dumps({"error": "只接受 POST 请求"})
        }

    # 解析请求体
    try:
        data = json.loads(event.get('body') or '{}')
    except json.JSONDecodeError:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": "请求体 JSON 格式不对"})
        }

    name = data.get('name', '匿名')
    age = data.get('age', 0)

    return {
        "statusCode": 200,
        "headers": {"content-type": "application/json; charset=utf-8"},
        "body": json.dumps({
            "message": f"收到数据, 名字: {name}, 年龄: {age}"
        }, ensure_ascii=False)
    }

路径参数

# index.py
import json

def main_handler(event, context):
    # 路径参数在 pathParameters 里
    path_params = event.get('pathParameters', {}) or {}
    user_id = path_params.get('id')

    if not user_id:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": "缺少用户 ID"})
        }

    return {
        "statusCode": 200,
        "headers": {"content-type": "application/json; charset=utf-8"},
        "body": json.dumps({
            "userId": user_id,
            "name": f"用户{user_id}"
        }, ensure_ascii=False)
    }

依赖管理

使用 requirements.txt

# requirements.txt
requests==2.31.0
python-dateutil==2.8.2

部署时平台会根据 requirements.txt 自动执行 pip install。你也可以本地先装好依赖测试通过后再部署。

注意: 如果依赖包含 C 扩展(比如 numpy)，要确保和本地环境跟云端的 Python 版本和操作系统一致，否则可能在云端跑不起来。

使用标准库

Python 自带的模块不需要安装，直接 import 就行。

# index.py
import json
import hashlib
import time

def main_handler(event, context):
    # 用 hashlib 生成哈希
    raw = f"hello-{time.time()}"
    token = hashlib.sha256(raw.encode()).hexdigest()

    return {
        "statusCode": 200,
        "headers": {"content-type": "application/json; charset=utf-8"},
        "body": json.dumps({"token": token})
    }

环境变量

在控制台配置的环境变量通过 os.environ 读取。

# index.py
import os
import json

def main_handler(event, context):
    # 读取环境变量
    api_key = os.environ.get('API_KEY', '')
    db_host = os.environ.get('DB_HOST', '')

    if not api_key:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": "缺少 API_KEY 环境变量"})
        }

    return {
        "statusCode": 200,
        "headers": {"content-type": "application/json; charset=utf-8"},
        "body": json.dumps({
            "dbHost": db_host or '未配置',
            "apiKeySet": True
        })
    }

完整示例: 待办事项 API

下面用 Python 写一个待办事项接口，功能和 Node.js 版本一样。

# index.py
import json
from datetime import datetime

# 内存中的待办列表(演示用，实际应该用数据库)
todos = []
next_id = 1

def main_handler(event, context):
    global next_id

    method = event.get('httpMethod', '')
    path = event.get('path', '')
    path_params = event.get('pathParameters', {}) or {}

    # 查询全部
    if path == '/api/todos' and method == 'GET':
        return {
            "statusCode": 200,
            "headers": {"content-type": "application/json; charset=utf-8"},
            "body": json.dumps({"data": todos}, ensure_ascii=False)
        }

    # 新增一条
    if path == '/api/todos' and method == 'POST':
        data = json.loads(event.get('body') or '{}')
        title = data.get('title', '').strip()

        if not title:
            return {
                "statusCode": 400,
                "body": json.dumps({"error": "title 不能为空"})
            }

        new_todo = {
            "id": next_id,
            "title": title,
            "done": False,
            "createdAt": datetime.now().isoformat()
        }
        next_id += 1
        todos.append(new_todo)

        return {
            "statusCode": 201,
            "headers": {"content-type": "application/json; charset=utf-8"},
            "body": json.dumps({"data": new_todo}, ensure_ascii=False)
        }

    # 删除一条
    if path.startswith('/api/todos/') and method == 'DELETE':
        todo_id = int(path_params.get('id', 0))
        target = None
        for todo in todos:
            if todo['id'] == todo_id:
                target = todo
                break

        if target is None:
            return {
                "statusCode": 404,
                "body": json.dumps({"error": "找不到这条待办"})
            }

        todos.remove(target)
        return {
            "statusCode": 200,
            "headers": {"content-type": "application/json; charset=utf-8"},
            "body": json.dumps({"message": "删除成功"})
        }

    return {
        "statusCode": 404,
        "body": json.dumps({"error": "路由不存在"})
    }

Python 和 Node.js 写法对比

Python 写法更直观，不需要到处写 await。但如果要做高并发的 I/O 操作(比如同时请求多个外部 API)，Node.js 的 async/await 用起来更自然。

速查卡片