从零构建无服务器 API：华为云 FunctionGraph + API Gateway 实战指南

一、为什么选择无服务器？

传统 Web 应用开发中，我们总是绕不开几个问题：

• 服务器运维：买机器、装系统、打补丁、配防火墙
• 弹性伸缩：流量低时空转，流量高峰时又来不及扩容
• 按需付费：包年包月模式下，闲置资源成本居高不下

无服务器架构（Serverless）正是为解决这些痛点而生。华为云 FunctionGraph 提供事件驱动的函数计算服务，API Gateway 提供 API 发布与管理能力，两者结合，让你只需关注业务代码本身。

本文将手把手带你从零搭建一个完整的无服务器 API 服务。

二、核心概念速览

三、实战：构建一个 Todo 后端 API

3.1 架构设计

客户端 → API Gateway → FunctionGraph → 华为云 GaussDB(for MySQL)

我们构建一个简单的 Todo 待办事项 API，包含以下接口：

3.2 步骤一：创建 GaussDB 数据库

1. 登录华为云控制台，进入 云数据库 GaussDB
2. 点击「创建数据库实例」，选择 GaussDB(for MySQL)
3. 配置参数：
   • 引擎版本：MySQL 8.0
   • 规格：推荐 2核4G（开发测试）
   • 存储：100GB
   • 虚拟私有云：选择已有 VPC 或新建
4. 创建完成后，进入实例详情，获取内网地址
5. 创建数据库 todo_db 和用户 todo_user

执行建表 SQL：

CREATE TABLE todos (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    title VARCHAR(255) NOT NULL,
    completed TINYINT DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

3.3 步骤二：创建 FunctionGraph 函数

1. 进入 FunctionGraph 控制台
2. 点击「创建函数」，选择 Python 3.9 运行时
3. 函数名称：todo-handler
4. 代码类型：在线编辑

完整函数代码如下：

import json
import os
import pymysql
from datetime import datetime

# 数据库连接配置（生产环境请使用华为云 DEW 凭据管理服务）
DB_HOST = os.environ.get('DB_HOST', 'your-db-host')
DB_PORT = int(os.environ.get('DB_PORT', '3306'))
DB_USER = os.environ.get('DB_USER', 'todo_user')
DB_PASSWORD = os.environ.get('DB_PASSWORD', 'your-password')
DB_NAME = os.environ.get('DB_NAME', 'todo_db')


def get_db_connection():
    """获取数据库连接"""
    return pymysql.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        database=DB_NAME,
        charset='utf8mb4'
    )


def parse_path(event):
    """解析请求路径和 ID"""
    path = event.get('path', '')
    parts = path.strip('/').split('/')
    return parts


def handler(event, context):
    """函数主入口"""
    http_method = event.get('httpMethod', 'GET')
    path_parts = parse_path(event)
    
    # CORS 预检请求
    if http_method == 'OPTIONS':
        return {
            'statusCode': 200,
            'headers': {
                'Access-Control-Allow-Origin': '*',
                'Access-Control-Allow-Methods': 'GET,POST,PUT,DELETE,OPTIONS',
                'Access-Control-Allow-Headers': 'Content-Type,Authorization'
            },
            'body': json.dumps({'message': 'OK'})
        }
    
    try:
        conn = get_db_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # GET /todos - 查询列表
        if http_method == 'GET' and len(path_parts) == 1 and path_parts[0] == 'todos':
            cursor.execute("SELECT * FROM todos ORDER BY created_at DESC")
            todos = cursor.fetchall()
            return {
                'statusCode': 200,
                'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                'body': json.dumps({'data': todos, 'code': 0})
            }
        
        # GET /todos/{id} - 查询详情
        elif http_method == 'GET' and len(path_parts) == 2:
            todo_id = path_parts[1]
            cursor.execute("SELECT * FROM todos WHERE id = %s", (todo_id,))
            todo = cursor.fetchone()
            if todo:
                return {
                    'statusCode': 200,
                    'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                    'body': json.dumps({'data': todo, 'code': 0})
                }
            else:
                return {
                    'statusCode': 404,
                    'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                    'body': json.dumps({'message': '未找到该待办', 'code': -1})
                }
        
        # POST /todos - 创建待办
        elif http_method == 'POST' and len(path_parts) == 1 and path_parts[0] == 'todos':
            body = json.loads(event.get('body', '{}'))
            title = body.get('title', '')
            if not title:
                return {
                    'statusCode': 400,
                    'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                    'body': json.dumps({'message': '标题不能为空', 'code': -1})
                }
            cursor.execute("INSERT INTO todos (title) VALUES (%s)", (title,))
            conn.commit()
            new_id = cursor.lastrowid
            cursor.execute("SELECT * FROM todos WHERE id = %s", (new_id,))
            todo = cursor.fetchone()
            return {
                'statusCode': 201,
                'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                'body': json.dumps({'data': todo, 'code': 0})
            }
        
        # PUT /todos/{id} - 更新状态
        elif http_method == 'PUT' and len(path_parts) == 2:
            todo_id = path_parts[1]
            body = json.loads(event.get('body', '{}'))
            completed = body.get('completed', 0)
            cursor.execute("UPDATE todos SET completed = %s WHERE id = %s", (completed, todo_id))
            conn.commit()
            if cursor.rowcount > 0:
                cursor.execute("SELECT * FROM todos WHERE id = %s", (todo_id,))
                todo = cursor.fetchone()
                return {
                    'statusCode': 200,
                    'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                    'body': json.dumps({'data': todo, 'code': 0})
                }
            else:
                return {
                    'statusCode': 404,
                    'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                    'body': json.dumps({'message': '未找到该待办', 'code': -1})
                }
        
        # DELETE /todos/{id} - 删除待办
        elif http_method == 'DELETE' and len(path_parts) == 2:
            todo_id = path_parts[1]
            cursor.execute("DELETE FROM todos WHERE id = %s", (todo_id,))
            conn.commit()
            if cursor.rowcount > 0:
                return {
                    'statusCode': 200,
                    'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                    'body': json.dumps({'message': '删除成功', 'code': 0})
                }
            else:
                return {
                    'statusCode': 404,
                    'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                    'body': json.dumps({'message': '未找到该待办', 'code': -1})
                }
        
        else:
            return {
                'statusCode': 404,
                'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
                'body': json.dumps({'message': '接口不存在', 'code': -1})
            }
    
    except Exception as e:
        return {
            'statusCode': 500,
            'headers': {'Access-Control-Allow-Origin': '*', 'Content-Type': 'application/json'},
            'body': json.dumps({'message': str(e), 'code': -1})
        }
    finally:
        try:
            cursor.close()
            conn.close()
        except:
            pass

关键配置说明：

• 在函数「配置」页签中，添加环境变量：

  • DB_HOST：GaussDB 内网地址
  • DB_PORT：3306
  • DB_USER：todo_user
  • DB_PASSWORD：数据库密码
  • DB_NAME：todo_db

• 在函数「依赖」页签中，添加依赖包：pymysql

3.4 步骤三：配置 VPC 访问

由于函数需要访问 GaussDB（在 VPC 内），必须配置 VPC 通道：

1. 进入 虚拟私有云 VPC 控制台
2. 确认函数与数据库在同一 VPC 下
3. 在函数「网络配置」中，选择对应 VPC 和子网
4. 确保安全组规则允许函数访问数据库端口（3306）

3.5 步骤四：创建 API Gateway 触发器

1. 在函数「触发器」页签，点击「创建触发器」
2. 触发器类型：API 网关触发器
3. 配置参数：
   • 网关类型：共享版（开发测试）/ 专享版（生产环境）
   • 部署环境：RELEASE（发布环境）
   • 认证方式：无认证（演示用，生产请使用 API 网关鉴权）
   • 请求路径：/todos
   • 请求方法：ANY（支持所有方法）
4. 创建完成后，获取 API 调用地址

3.6 步骤五：测试 API

使用 curl 或 Postman 测试：

# 创建待办
curl -X POST https://{api-gateway-url}/todos \
  -H "Content-Type: application/json" \
  -d '{"title": "学习华为云 FunctionGraph"}'

# 查询列表
curl -X GET https://{api-gateway-url}/todos

# 更新状态
curl -X PUT https://{api-gateway-url}/todos/1 \
  -H "Content-Type: application/json" \
  -d '{"completed": 1}'

# 删除待办
curl -X DELETE https://{api-gateway-url}/todos/1

四、生产环境优化建议

4.1 安全加固

• API 网关鉴权：使用华为云 IAM 鉴权或 HMAC 签名认证
• DEW 凭据管理：不要将数据库密码硬编码在环境变量中，使用 DEW 凭据管理服务
• WAF 防护：为 API 网关绑定 Web 应用防火墙，防止 SQL 注入、XSS 等攻击

4.2 性能优化

• 函数复用：启用函数复用，减少冷启动时间
• 连接池：在函数初始化阶段建立数据库连接池，避免每次调用都新建连接
• 异步处理：对于耗时操作，使用函数异步调用 + 消息通知

4.3 可观测性

• APM 应用性能监控：在函数中集成 APM SDK，追踪调用链路
• LTS 日志服务：函数日志自动接入 LTS，方便排查问题
• 告警配置：为函数错误率、调用延迟配置告警规则

4.4 成本优化

• 函数规格：根据实际负载选择合适的函数规格（内存、CPU）
• 预留模式：对于有稳定基线流量的场景，使用预留模式避免冷启动
• API 网关限流：配置限流策略，防止突发流量产生高额费用

五、总结

通过本文，我们完成了：

✅ 使用 GaussDB(for MySQL) 作为数据层
✅ 使用 FunctionGraph 编写业务逻辑
✅ 使用 API Gateway 暴露 RESTful API
✅ 配置 VPC 通道实现内网安全访问

整个过程零服务器运维，代码部署后即可自动扩缩容，按调用量计费。对于初创团队、个人开发者、以及需要快速验证业务场景的团队来说，无服务器架构是一个极具性价比的选择。

华为云 FunctionGraph 还支持与 OBS（对象存储）、DMS（消息队列）、SMN（消息通知） 等多种服务联动，可以构建更复杂的业务场景。后续文章我们将探索如何用函数构建事件驱动的微服务架构。

参考资料

• 华为云 FunctionGraph 官方文档
• API Gateway 使用指南
• GaussDB(for MySQL) 开发指南
• 无服务器架构最佳实践

声明：本文为华为云开发者社区原创内容，代码示例仅供学习参考。生产环境请遵循华为云安全最佳实践，合理配置鉴权、限流、监控等策略。