API 开发

从命令行构建、测试、文档化和调试 HTTP API。覆盖完整的 API 生命周期：端点脚手架搭建、使用 curl 测试、生成 OpenAPI 文档、模拟服务以及调试。

适用场景

• 搭建新的 REST 或 GraphQL 端点
• 使用 curl 或脚本测试 API
• 生成或验证 OpenAPI/Swagger 规范
• 为开发模拟外部 API
• 调试 HTTP 请求/响应问题
• 对端点进行负载测试

使用 curl 测试 API

GET 请求

# 基础 GET
curl -s https://api.example.com/users | jq .

# 带请求头
curl -s -H "Authorization: Bearer $TOKEN" 
  -H "Accept: application/json" 
  https://api.example.com/users | jq .

# 带查询参数
curl -s "https://api.example.com/users?page=2&limit=10" | jq .

# 同时显示响应头
curl -si https://api.example.com/users

POST/PUT/PATCH/DELETE

# POST JSON
curl -s -X POST https://api.example.com/users 
  -H "Content-Type: application/json" 
  -H "Authorization: Bearer $TOKEN" 
  -d '{"name": "Alice", "email": "alice@example.com"}' | jq .

# PUT（全量替换）
curl -s -X PUT https://api.example.com/users/123 
  -H "Content-Type: application/json" 
  -d '{"name": "Alice Updated", "email": "alice@example.com"}' | jq .

# PATCH（部分更新）
curl -s -X PATCH https://api.example.com/users/123 
  -H "Content-Type: application/json" 
  -d '{"name": "Alice V2"}' | jq .

# DELETE
curl -s -X DELETE https://api.example.com/users/123

# POST 表单数据
curl -s -X POST https://api.example.com/upload 
  -F "file=@document.pdf" 
  -F "description=My document"

调试请求

# 详细输出（查看完整请求/响应）
curl -v https://api.example.com/health 2>&1

# 仅显示响应头
curl -sI https://api.example.com/health

# 显示耗时分解
curl -s -o /dev/null -w "DNS: %{time_namelookup}snConnect: %{time_connect}snTLS: %{time_appconnect}snFirst byte: %{time_starttransfer}snTotal: %{time_total}sn" https://api.example.com/health

# 跟随重定向
curl -sL https://api.example.com/old-endpoint

# 将响应保存到文件
curl -s -o response.json https://api.example.com/data

API 测试脚本

Bash 测试运行器

#!/bin/bash
# api-test.sh - 简单的 API 测试运行器
BASE_URL="${1:-http://localhost:3000}"
PASS=0
FAIL=0

assert_status() {
  local method="$1" url="$2" expected="$3" body="$4"
  local args=(-s -o /dev/null -w "%{http_code}" -X "$method")
  if [ -n "$body" ]; then
    args+=(-H "Content-Type: application/json" -d "$body")
  fi
  local status
  status=$(curl "${args[@]}" "$BASE_URL$url")
  if [ "$status" = "$expected" ]; then
    echo "PASS: $method $url -> $status"
    ((PASS++))
  else
    echo "FAIL: $method $url -> $status (expected $expected)"
    ((FAIL++))
  fi
}

assert_json() {
  local url="$1" jq_expr="$2" expected="$3"
  local actual
  actual=$(curl -s "$BASE_URL$url" | jq -r "$jq_expr")
  if [ "$actual" = "$expected" ]; then
    echo "PASS: GET $url | jq '$jq_expr' = $expected"
    ((PASS++))
  else
    echo "FAIL: GET $url | jq '$jq_expr' = $actual (expected $expected)"
    ((FAIL++))
  fi
}

# 健康检查
assert_status GET /health 200

# CRUD 测试
assert_status POST /api/users 201 '{"name":"Test","email":"test@test.com"}'
assert_status GET /api/users 200
assert_json /api/users '.[-1].name' 'Test'
assert_status DELETE /api/users/1 204

# 认证测试
assert_status GET /api/admin 401
assert_status GET /api/admin 403  # 使用错误角色

echo ""
echo "结果：$PASS 通过，$FAIL 失败"
[ "$FAIL" -eq 0 ] && exit 0 || exit 1

Python 测试运行器

#!/usr/bin/env python3
"""api_test.py - API 集成测试套件。"""
import json, sys, urllib.request, urllib.error

BASE = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:3000"
PASS = FAIL = 0

def request(method, path, body=None, headers=None):
    """发起 HTTP 请求，返回 (状态码, 响应体字典, 响应头)。"""
    url = f"{BASE}{path}"
    data = json.dumps(body).encode() if body else None
    hdrs = {"Content-Type": "application/json", "Accept": "application/json"}
    if headers:
        hdrs.update(headers)
    req = urllib.request.Request(url, data=data, headers=hdrs, method=method)
    try:
        resp = urllib.request.urlopen(req)
        body = json.loads(resp.read().decode()) if resp.read() else None
    except urllib.error.HTTPError as e:
        return e.code, None, dict(e.headers)
    return resp.status, body, dict(resp.headers)

def test(name, fn):
    """运行测试函数，并统计通过/失败数。"""
    global PASS, FAIL
    try:
        fn()
        print(f"  PASS: {name}")
        PASS += 1
    except AssertionError as e:
        print(f"  FAIL: {name} - {e}")
        FAIL += 1

def assert_eq(actual, expected, msg=""):
    assert actual == expected, f"得到 {actual}，期望 {expected}。{msg}"

# --- 测试用例 ---
print(f"正在测试 {BASE}n")

test("GET /health 返回 200", lambda: (
    assert_eq(request("GET", "/health")[0], 200)
))

test("POST /api/users 创建用户", lambda: (
    assert_eq(request("POST", "/api/users", {"name": "Test", "email": "t@t.com"})[0], 201)
))

test("GET /api/users 返回数组", lambda: (
    assert_eq(type(request("GET", "/api/users")[1]), list)
))

test("GET /api/notfound 返回 404", lambda: (
    assert_eq(request("GET", "/api/notfound")[0], 404)
))

print(f"n结果：{PASS} 通过，{FAIL} 失败")
sys.exit(0 if FAIL == 0 else 1)

OpenAPI 规范生成

从现有端点生成

# 从 curl 响应中搭建 OpenAPI 3.0 规范
# 运行此命令后，再补充细节
cat > openapi.yaml << 'EOF'
openapi: "3.0.3"
info:
  title: My API
  version: "1.0.0"
  description: API description here
servers:
  - url: http://localhost:3000
    description: Local development
paths:
  /health:
    get:
      summary: Health check
      responses:
        "200":
          description: Service is healthy
          content:
            application/json:
              schema:
                type: object
                properties:
                  status:
                    type: string
                    example: ok
  /api/users:
    get:
      summary: List users
      parameters:
        - name: page
          in: query
          schema:
            type: integer
            default: 1
        - name: limit
          in: query
          schema:
            type: integer
            default: 20
      responses:
        "200":
          description: List of users
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: "#/components/schemas/User"
    post:
      summary: Create user
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: "#/components/schemas/CreateUser"
      responses:
        "201":
          description: User created
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/User"
        "400":
          description: Validation error
  /api/users/{id}:
    get:
      summary: Get user by ID
      parameters:
        - name: id
          in: path
          required: true
          schema:
            type: string
      responses:
        "200":
          description: User details
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/User"
        "404":
          description: Not found
components:
  schemas:
    User:
      type: object
      properties:
        id:
          type: string
        name:
          type: string
        email:
          type: string
          format: email
        createdAt:
          type: string
          format: date-time
    CreateUser:
      type: object
      required:
        - name
        - email
      properties:
        name:
          type: string
        email:
          type: string
          format: email
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT
EOF

验证 OpenAPI 规范

# 使用 npx（无需安装）
npx @redocly/cli lint openapi.yaml

# 快速检查：YAML 是否有效？
python3 -c "import yaml; yaml.safe_load(open('openapi.yaml'))" && echo "Valid YAML"

模拟服务器

使用 Python 快速启动模拟服务

#!/usr/bin/env python3
"""mock_server.py - 基于类 OpenAPI 配置的轻量级 API 模拟服务。"""
import json, http.server, re, sys

PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8080

# 定义模拟路由：(方法, 路径模式) -> 响应
ROUTES = {
    ("GET", "/health"): {"status": 200, "body": {"status": "ok"}},
    ("GET", "/api/users"): {"status": 200, "body": [
        {"id": "1", "name": "Alice", "email": "alice@example.com"},
        {"id": "2", "name": "Bob", "email": "bob@example.com"},
    ]},
    ("POST", "/api/users"): {"status": 201, "body": {"id": "3", "name": "Created"}},
    ("GET", r"/api/users/w+"): {"status": 200, "body": {"id": "1", "name": "Alice"}},
    ("DELETE", r"/api/users/w+"): {"status": 204, "body": None},
}

class MockHandler(http.server.BaseHTTPRequestHandler):
    def _handle(self):
        for (method, pattern), response in ROUTES.items():
            if self.command == method and re.fullmatch(pattern, self.path.split('?')[0]):
                self.send_response(response["status"])
                if response["body"] is not None:
                    self.send_header("Content-Type", "application/json")
                    self.end_headers()
                    self.wfile.write(json.dumps(response["body"]).encode())
                else:
                    self.end_headers()
                return
        self.send_response(404)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps({"error": "Not found"}).encode())

    do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = _handle

    def log_message(self, fmt, *args):
        print(f"{self.command} {self.path} -> {args[1] if len(args) > 1 else '?'}")

print(f"模拟服务器运行在 http://localhost:{PORT}")
http.server.HTTPServer(("", PORT), MockHandler).serve_forever()

运行：python3 mock_server.py 8080

Node.js Express 脚手架

最小化 REST API

// server.js - 最小化的 Express REST API
const express = require('express');
const app = express();
app.use(express.json());

// 内存存储
const items = new Map();
let nextId = 1;

// CRUD 端点
app.get('/api/items', (req, res) => {
  const { page = 1, limit = 20 } = req.query;
  const all = [...items.values()];
  const start = (page - 1) * limit;
  res.json({ items: all.slice(start, start + +limit), total: all.length });
});

app.get('/api/items/:id', (req, res) => {
  const item = items.get(req.params.id);
  if (!item) return res.status(404).json({ error: 'Not found' });
  res.json(item);
});

app.post('/api/items', (req, res) => {
  const { name, description } = req.body;
  if (!name) return res.status(400).json({ error: 'name required' });
  const id = String(nextId++);
  const item = { id, name, description: description || '', createdAt: new Date().toISOString() };
  items.set(id, item);
  res.status(201).json(item);
});

app.put('/api/items/:id', (req, res) => {
  if (!items.has(req.params.id)) return res.status(404).json({ error: 'Not found' });
  const item = { ...req.body, id: req.params.id, updatedAt: new Date().toISOString() };
  items.set(req.params.id, item);
  res.json(item);
});

app.delete('/api/items/:id', (req, res) => {
  if (!items.has(req.params.id)) return res.status(404).json({ error: 'Not found' });
  items.delete(req.params.id);
  res.status(204).end();
});

// 错误处理器
app.use((err, req, res, next) => {
  console.error(err.stack);
  res.status(500).json({ error: 'Internal server error' });
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => console.log(`API running on http://localhost:${PORT}`));

初始化步骤

mkdir my-api && cd my-api
npm init -y
npm install express
node server.js

调试技巧

检查端口是否被占用

# Linux/macOS
lsof -i :3000
# 或
ss -tlnp | grep 3000

# 终止占用该端口的进程
kill $(lsof -t -i :3000)

测试 CORS

# 预检请求
curl -s -X OPTIONS https://api.example.com/users 
  -H "Origin: http://localhost:3000" 
  -H "Access-Control-Request-Method: POST" 
  -H "Access-Control-Request-Headers: Content-Type" 
  -I

监控响应时间退化

# 快速基准测试（10 次请求）
for i in $(seq 1 10); do
  curl -s -o /dev/null -w "%{time_total}n" http://localhost:3000/api/users
done | awk '{sum+=$1; if($1>max)max=$1} END {printf "Avg: %.3fs, Max: %.3fsn", sum/NR, max}'

解析 JWT 令牌

# 解码 JWT 载荷（不验证签名）
echo "$TOKEN" | cut -d. -f2 | base64 -d 2>/dev/null | jq .

提示

• 使用 jq 处理 JSON 响应：curl -s url | jq '.items[] | {id, name}'
• 所有含请求体的请求都应设置 Content-Type 请求头 —— 缺失会导致静默的 400 错误
• 使用 curl 的 -w 'n' 参数确保输出以换行符结尾
• 对于大响应体，可管道至 jq -C . | less -R 实现带颜色的分页浏览
• 务必测试错误路径：无效 JSON、缺失字段、类型错误、未授权、资源未找到等
• WebSocket 测试：npx wscat -c ws://localhost:3000/ws