FunctionGraph 实战：Python 写 HTTP 接口与 OBS 触发器及避坑指南

前言

Serverless 最吸引人的地方，是“只写业务代码，不管服务器”。但真正把函数放到生产环境，很快就会遇到一串问题：第一次请求为什么这么慢？明明本地能跑，上传后为什么报 ModuleNotFoundError？函数里访问 OBS 为什么提示无权限？

本文以华为云函数工作流 FunctionGraph 为例，用 Python 完成两个最常见的场景：

1. 通过 APIG 触发器对外提供一个 HTTP 接口；
2. 通过 OBS 触发器，在文件上传后自动处理对象。

最后把冷启动、超时、依赖包、权限委托这几类高频问题集中梳理一遍。文中涉及的配额、默认值和计费规则会随版本调整，请以控制台和官方文档为准。

一、准备工作

• 一个已实名的华为云账号，并选定区域（下文以 cn-north-4 为例）；
• 创建一个 OBS 桶用于存放上传文件，再建一个桶（或独立前缀）存放处理结果；
• 本地准备与函数运行时一致的 Python 版本，便于打包依赖。

在 FunctionGraph 控制台创建函数时，选择“创建空白函数”，运行时选择 Python 3.x，函数执行入口填写 index.handler，表示调用 index.py 中的 handler 函数。

二、场景一：用 APIG 触发器写一个 HTTP 接口

2.1 函数代码

APIG 触发器会把 HTTP 请求封装成一个 event 字典传给函数，函数需要按约定的结构返回响应。

# index.py
import json
import base64

def handler(event, context):
    logger = context.getLogger()
    logger.info("request id: %s", context.getRequestID())

    method = event.get("httpMethod", "GET")
    query = event.get("queryStringParameters") or {}

    body = event.get("body") or ""
    # APIG 可能对请求体做 Base64 编码，务必先判断
    if body and event.get("isBase64Encoded"):
        body = base64.b64decode(body).decode("utf-8")

    data = {}
    if method == "POST" and body:
        try:
            data = json.loads(body)
        except ValueError:
            return _resp(400, {"error": "body must be JSON"})

    name = data.get("name") or query.get("name") or "FunctionGraph"
    return _resp(200, {"message": f"Hello, {name}!", "method": method})


def _resp(status, payload):
    return {
        "statusCode": status,
        "isBase64Encoded": False,
        "headers": {"Content-Type": "application/json; charset=utf-8"},
        "body": json.dumps(payload, ensure_ascii=False),
    }

2.2 创建触发器

在函数详情页选择“创建触发器”，类型选 API 网关服务（APIG），安全认证在调试阶段可以选“None”，上线前务必改为 App 认证或 IAM 认证。创建后控制台会给出调用地址，用 curl 测试：

curl -X POST "<你的API调用地址>" \
  -H "Content-Type: application/json" \
  -d '{"name": "silzhan"}'

2.3 要点

• 返回值必须包含 statusCode、body 等字段，body 必须是字符串，直接返回字典会导致网关报错；
• 记得处理 isBase64Encoded，这是最常见的“本地正常、线上解析失败”的原因；
• 如果你已有 Flask、FastAPI 等 Web 框架代码，也可以了解 FunctionGraph 的 HTTP 函数类型，它以监听端口的方式运行完整 Web 服务，具体启动方式和端口要求请参考官方文档。

三、场景二：OBS 触发器自动处理上传文件

3.1 函数代码

需求：用户把文本文件上传到 upload/ 前缀下，函数统计行数，并把结果写入另一个桶。访问 OBS 使用官方 SDK esdk-obs-python，导入方式为 from obs import ObsClient。

# index.py
import os
import json
from urllib.parse import unquote_plus
from obs import ObsClient

_client = None  # 放在全局，热启动时可复用

def _get_client(context):
    global _client
    if _client is None:
        _client = ObsClient(
            access_key_id=context.getSecurityAccessKey(),
            secret_access_key=context.getSecuritySecretKey(),
            security_token=context.getSecurityToken(),
            server=os.environ.get("OBS_ENDPOINT"),
        )
    return _client

def handler(event, context):
    logger = context.getLogger()
    client = _get_client(context)
    output_bucket = os.environ.get("OUTPUT_BUCKET")

    for record in event.get("Records", []):
        info = record.get("s3") or record.get("obs") or {}
        bucket = info["bucket"]["name"]
        # 对象名在事件里是 URL 编码的，中文和空格必须解码
        key = unquote_plus(info["object"]["key"])
        logger.info("new object: %s/%s", bucket, key)

        resp = client.getObject(bucket, key, loadStreamInMemory=True)
        if resp.status >= 300:
            logger.error("get failed: %s", resp.errorMessage)
            continue

        content = resp.body.buffer.decode("utf-8", errors="ignore")
        result = {"key": key, "lines": content.count("\n") + 1}

        client.putContent(output_bucket, f"result/{key}.json",
                          json.dumps(result, ensure_ascii=False))
    return {"processed": len(event.get("Records", []))}

在“设置 → 环境变量”中配置 OBS_ENDPOINT（例如 obs.cn-north-4.myhuaweicloud.com）和 OUTPUT_BUCKET。

3.2 创建触发器

触发器类型选择“对象存储服务（OBS）”，选中源桶，事件选择对象创建相关事件，并设置前缀 upload/、后缀 .txt。上传一个文件后，到“监控 → 日志”中确认执行结果。

3.3 小心递归触发

如果把结果写回源桶，而且没有设置前缀/后缀过滤，写入动作会再次触发函数，形成死循环，既浪费资源又可能产生额外费用。最稳妥的做法是：输出写到另一个桶；至少也要保证输出路径不匹配触发器的过滤规则。

四、常见坑与解决方案

4.1 冷启动

函数实例空闲一段时间后会被回收，下一次请求需要重新拉起实例、加载代码和依赖，这就是冷启动。优化思路：

• 减小代码包：删掉测试文件、文档和无用依赖；
• 全局初始化：数据库连接、SDK 客户端放在 handler 之外，热实例可复用；
• 使用初始化函数：在控制台配置初始化入口（如 index.initializer），把耗时的准备工作和请求处理分开；
• 预留实例：对延迟敏感的接口，可以配置预留实例保持常驻，注意它会产生持续费用，计费方式以官方说明为准。

def initializer(context):
    # 只在实例启动时执行一次
    global MODEL
    MODEL = load_something_heavy()

4.2 超时

函数的执行超时时间可在“设置 → 常规设置”中调整，新建函数的默认值通常较短，处理大文件时很容易超时。注意两点：

• APIG 后端本身也有超时限制，HTTP 接口的实际可用时长受两者共同约束，具体上限请查看控制台；
• 长任务建议改为异步：接口只负责接收请求、写入队列或 OBS，再由其他函数处理。代码里可以用 context.getRemainingTimeInMilliSeconds() 判断剩余时间，提前收尾。

4.3 依赖包

ModuleNotFoundError 几乎是每个人的必经之路。函数运行在 Linux 环境中，因此：

# 在 Linux 或与运行时一致的容器里安装
mkdir package && cd package
pip install esdk-obs-python -t .
cp ../index.py .
zip -r ../function.zip .

• 依赖要和代码放在 zip 的根目录，不要多套一层文件夹；
• 带 C 扩展的库（如 Pillow、numpy）必须在 Linux、相同 Python 版本下编译，Windows/macOS 上装的包无法使用；
• 多个函数共用的依赖，可以上传到“依赖包管理”中，再在函数里引用，减小每个函数的代码包；代码包和依赖包的大小上限请以官方文档为准。

4.4 权限委托

函数代码里不应该写死 AK/SK。正确做法是在 IAM 中创建委托，委托对象选择云服务 FunctionGraph，授予所需的 OBS 权限，再在函数“权限”配置中选择该委托。之后通过 context.getSecurityAccessKey()、getSecuritySecretKey()、getSecurityToken() 获取临时凭证。

常见错误：

• 未配置委托，context 拿到的临时凭证为空，访问 OBS 返回 403；
• 使用临时凭证但忘记传 security_token，同样鉴权失败；
• 授权范围过大，建议按最小权限原则，只授予指定桶的读写权限；
• 临时凭证有有效期，如果全局缓存客户端，要考虑凭证过期后重建客户端。

4.5 日志与调试

排查问题时，日志是最直接的依据。建议统一使用 context.getLogger() 输出日志，并在每条关键日志中带上请求 ID，方便在多次调用中定位同一请求。函数日志可以在函数详情页直接查看，也可以对接云日志服务 LTS 做长期留存和检索，留存时长和计费请以官方说明为准。

调试阶段可以充分利用控制台的“配置测试事件”功能：平台提供了 APIG、OBS 等触发器的事件模板，把模板中的桶名、对象名改成真实值，就能在不真正上传文件的情况下验证代码逻辑。需要注意的是，测试事件中的字段结构以模板为准，不同版本可能存在差异，代码里对字段读取尽量做好容错，例如使用 dict.get() 而不是直接下标访问。

另外，环境变量适合存放桶名、终端节点等配置，但密码、密钥类信息建议开启加密或使用专门的凭据管理服务，不要明文写进代码仓库。

五、总结

FunctionGraph 让我们用很少的代码就能搭建接口和事件驱动的数据处理流程，但把这些坑提前想清楚，才能让函数在生产环境中稳定运行。