如何搭建Serverless Framework+TS对接AWS多服务的极简数据读写项目

项目结构 #

├── src/
│   ├── handlers/
│   │   ├── getItems.ts    # GET接口处理函数
│   │   └── postItem.ts    # POST接口处理函数
│   └── config/
│       └── knex.ts        # Knex数据库连接配置
├── serverless.yml         # Serverless部署配置
├── tsconfig.json          # TypeScript配置
├── package.json           # 依赖声明
└── .env                   # 环境变量（禁止提交到代码仓库）

第一步：安装依赖 #

# 生产依赖
npm install knex mysql2 aws-sdk @types/aws-lambda
# 开发依赖
npm install -D serverless serverless-offline serverless-plugin-typescript typescript @types/node dotenv

第二步：核心配置文件 #

tsconfig.json #

{
  "compilerOptions": {
    "target": "ES2020",
    "module": "CommonJS",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules"]
}

.env 环境变量示例 #

RDS_HOST=你的RDS实例地址
RDS_USER=RDS登录用户名
RDS_PASSWORD=RDS登录密码
RDS_DATABASE=目标数据库名称
RDS_PORT=3306
COGNITO_USER_POOL_ARN=你的Cognito用户池ARN
VPC_SECURITY_GROUP_ID=RDS所属安全组ID
VPC_SUBNET_ID_1=VPC私有子网ID1
VPC_SUBNET_ID_2=VPC私有子网ID2

src/config/knex.ts 数据库配置 #

import knex from 'knex';
import * as dotenv from 'dotenv';
dotenv.config();

const db = knex({
  client: 'mysql2',
  connection: {
    host: process.env.RDS_HOST,
    user: process.env.RDS_USER,
    password: process.env.RDS_PASSWORD,
    database: process.env.RDS_DATABASE,
    port: Number(process.env.RDS_PORT || 3306)
  }
});

export default db;

serverless.yml 部署配置 #

service: ts-serverless-mini-demo
frameworkVersion: '3'

plugins:
  - serverless-plugin-typescript
  - serverless-offline

provider:
  name: aws
  runtime: nodejs18.x
  region: us-east-1 # 替换为你需要的AWS区域
  environment:
    RDS_HOST: ${env:RDS_HOST}
    RDS_USER: ${env:RDS_USER}
    RDS_PASSWORD: ${env:RDS_PASSWORD}
    RDS_DATABASE: ${env:RDS_DATABASE}
    RDS_PORT: ${env:RDS_PORT}
  # Lambda VPC配置，确保和RDS在同一个VPC下可访问
  vpc:
    securityGroupIds:
      - ${env:VPC_SECURITY_GROUP_ID}
    subnetIds:
      - ${env:VPC_SUBNET_ID_1}
      - ${env:VPC_SUBNET_ID_2}

functions:
  getItems:
    handler: src/handlers/getItems.handler
    events:
      - http:
          path: items
          method: get
          cors: true
          authorizer:
            type: COGNITO_USER_POOLS
            authorizerId:
              Ref: ApiGatewayAuthorizer
  postItem:
    handler: src/handlers/postItem.handler
    events:
      - http:
          path: items
          method: post
          cors: true
          authorizer:
            type: COGNITO_USER_POOLS
            authorizerId:
              Ref: ApiGatewayAuthorizer

resources:
  Resources:
    # Cognito授权器配置
    ApiGatewayAuthorizer:
      Type: AWS::ApiGateway::Authorizer
      Properties:
        Name: CognitoAuthorizer
        Type: COGNITO_USER_POOLS
        IdentitySource: method.request.header.Authorization
        RestApiId:
          Ref: ApiGatewayRestApi
        ProviderARNs:
          - ${env:COGNITO_USER_POOL_ARN}

第三步：接口处理函数 #

GET接口 src/handlers/getItems.ts #

import { APIGatewayProxyEvent, APIGatewayProxyResult } from 'aws-lambda';
import db from '../config/knex';

export const handler = async (event: APIGatewayProxyEvent): Promise<APIGatewayProxyResult> => {
  try {
    // 示例查询items表所有数据，可根据业务修改
    const items = await db('items').select('*');
    return {
      statusCode: 200,
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(items)
    };
  } catch (error) {
    return {
      statusCode: 500,
      body: JSON.stringify({ message: '查询失败', error: (error as Error).message })
    };
  }
};

POST接口 src/handlers/postItem.ts #

import { APIGatewayProxyEvent, APIGatewayProxyResult } from 'aws-lambda';
import db from '../config/knex';

export const handler = async (event: APIGatewayProxyEvent): Promise<APIGatewayProxyResult> => {
  try {
    if (!event.body) {
      return { statusCode: 400, body: JSON.stringify({ message: '请求参数为空' }) };
    }
    const data = JSON.parse(event.body);
    // 示例向items表插入数据，可根据业务修改
    const [insertId] = await db('items').insert(data);
    return {
      statusCode: 201,
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ id: insertId, ...data })
    };
  } catch (error) {
    return {
      statusCode: 500,
      body: JSON.stringify({ message: '插入失败', error: (error as Error).message })
    };
  }
};

第四步：部署步骤 #

1. 前置准备：提前在AWS控制台创建RDS MySQL实例、Cognito用户池，确保RDS安全组允许Lambda所在VPC访问3306端口，本地配置好AWS访问凭证
2. 提前在RDS中创建业务数据表，示例中用到的items表可根据需求自定义字段
3. 本地测试：执行sls offline启动本地服务调试
4. 部署到AWS：执行sls deploy完成全量资源自动部署

注意事项 #

• 本实现仅保留核心逻辑，无多余扩展功能，可根据需求自行添加参数校验、错误告警等逻辑
• 生产环境建议将敏感信息存入AWS Secrets Manager，不要直接明文写在.env文件中

内容的提问来源于stack exchange，提问作者Far Zad