JSON API：构建和使用 REST 服务

# JSON API：构建和使用 REST 服务

REST（Representational State Transfer）API 使用 JSON 作为请求和响应主体的主要数据格式。本指南涵盖构建和使用现代 JSON API 所需的一切。

REST API 基础

什么是 REST？

REST 是一种架构风格，用于设计网络应用程序。它依赖于无状态、客户端-服务器通信，主要使用 HTTP。

为什么使用 JSON？

• ✅ 轻量级：比 XML 更小的有效载荷
• ✅ 易于解析：原生 JavaScript 支持
• ✅ 人类可读：易于调试
• ✅ 广泛支持：所有编程语言
• ✅ 灵活：处理复杂数据结构

HTTP 方法

| 方法 | 用途 | 示例 |

|------|------|------|

| GET | 检索数据 | 获取用户列表 |

| POST | 创建新资源 | 创建新用户 |

| PUT | 更新整个资源 | 替换用户 |

| PATCH | 部分更新 | 更新用户电子邮件 |

| DELETE | 删除资源 | 删除用户 |

RESTful 端点示例

GET    /api/users          # 获取所有用户
GET    /api/users/123      # 获取用户 123
POST   /api/users          # 创建新用户
PUT    /api/users/123      # 更新用户 123
PATCH  /api/users/123      # 部分更新用户 123
DELETE /api/users/123      # 删除用户 123

使用 Express.js 构建 API

基本设置

安装：

npm init -y
npm install express

基本服务器：

import express from 'express';

const app = express();
const PORT = 3000;

// 中间件解析 JSON
app.use(express.json());

// 内存存储（生产中使用数据库）
let users = [
  { id: 1, name: "张三", email: "zhangsan@example.com" },
  { id: 2, name: "李四", email: "lisi@example.com" }
];

// GET 所有用户
app.get('/api/users', (req, res) => {
  res.json({
    success: true,
    data: users,
    count: users.length
  });
});

// GET 单个用户
app.get('/api/users/:id', (req, res) => {
  const user = users.find(u => u.id === parseInt(req.params.id));

  if (!user) {
    return res.status(404).json({
      success: false,
      error: '用户未找到'
    });
  }

  res.json({
    success: true,
    data: user
  });
});

// POST 创建用户
app.post('/api/users', (req, res) => {
  const { name, email } = req.body;

  // 验证
  if (!name || !email) {
    return res.status(400).json({
      success: false,
      error: '姓名和电子邮件为必填项'
    });
  }

  const user = {
    id: users.length + 1,
    name,
    email,
    createdAt: new Date().toISOString()
  };

  users.push(user);

  res.status(201).json({
    success: true,
    data: user
  });
});

// PUT 更新用户
app.put('/api/users/:id', (req, res) => {
  const id = parseInt(req.params.id);
  const index = users.findIndex(u => u.id === id);

  if (index === -1) {
    return res.status(404).json({
      success: false,
      error: '用户未找到'
    });
  }

  const user = {
    id,
    ...req.body,
    updatedAt: new Date().toISOString()
  };

  users[index] = user;

  res.json({
    success: true,
    data: user
  });
});

// PATCH 部分更新
app.patch('/api/users/:id', (req, res) => {
  const id = parseInt(req.params.id);
  const user = users.find(u => u.id === id);

  if (!user) {
    return res.status(404).json({
      success: false,
      error: '用户未找到'
    });
  }

  Object.assign(user, req.body);
  user.updatedAt = new Date().toISOString();

  res.json({
    success: true,
    data: user
  });
});

// DELETE 删除用户
app.delete('/api/users/:id', (req, res) => {
  const id = parseInt(req.params.id);
  const index = users.findIndex(u => u.id === id);

  if (index === -1) {
    return res.status(404).json({
      success: false,
      error: '用户未找到'
    });
  }

  users.splice(index, 1);

  res.json({
    success: true,
    message: '用户已删除'
  });
});

// 启动服务器
app.listen(PORT, () => {
  console.log(服务器运行在 http://localhost:${PORT});
});

错误处理中间件

// 错误处理中间件
app.use((err, req, res, next) => {
  console.error(err.stack);

  res.status(err.status || 500).json({
    success: false,
    error: err.message || '服务器错误',
    ...(process.env.NODE_ENV === 'development' && { stack: err.stack })
  });
});

// 404 处理
app.use((req, res) => {
  res.status(404).json({
    success: false,
    error: '未找到路由'
  });
});

使用 Python FastAPI 构建 API

基本设置

安装：

pip install fastapi uvicorn pydantic

基本 API：

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, EmailStr
from typing import List, Optional
from datetime import datetime

app = FastAPI(title="用户 API", version="1.0.0")

# 数据模型
class UserCreate(BaseModel):
    name: str
    email: EmailStr

class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[EmailStr] = None

class User(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime
    updated_at: Optional[datetime] = None

# 内存存储
users_db = []
user_id_counter = 1

# GET 所有用户
@app.get("/api/users", response_model=List[User])
def get_users():
    return users_db

# GET 单个用户
@app.get("/api/users/{user_id}", response_model=User)
def get_user(user_id: int):
    user = next((u for u in users_db if u["id"] == user_id), None)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="用户未找到"
        )
    return user

# POST 创建用户
@app.post("/api/users", response_model=User, status_code=status.HTTP_201_CREATED)
def create_user(user_data: UserCreate):
    global user_id_counter

    # 检查电子邮件是否已存在
    if any(u["email"] == user_data.email for u in users_db):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="电子邮件已存在"
        )

    user = {
        "id": user_id_counter,
        "name": user_data.name,
        "email": user_data.email,
        "created_at": datetime.now(),
        "updated_at": None
    }

    users_db.append(user)
    user_id_counter += 1

    return user

# PUT 更新用户
@app.put("/api/users/{user_id}", response_model=User)
def update_user(user_id: int, user_data: UserCreate):
    user = next((u for u in users_db if u["id"] == user_id), None)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="用户未找到"
        )

    user["name"] = user_data.name
    user["email"] = user_data.email
    user["updated_at"] = datetime.now()

    return user

# PATCH 部分更新
@app.patch("/api/users/{user_id}", response_model=User)
def partial_update_user(user_id: int, user_data: UserUpdate):
    user = next((u for u in users_db if u["id"] == user_id), None)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="用户未找到"
        )

    if user_data.name is not None:
        user["name"] = user_data.name
    if user_data.email is not None:
        user["email"] = user_data.email

    user["updated_at"] = datetime.now()

    return user

# DELETE 删除用户
@app.delete("/api/users/{user_id}")
def delete_user(user_id: int):
    global users_db

    user = next((u for u in users_db if u["id"] == user_id), None)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="用户未找到"
        )

    users_db = [u for u in users_db if u["id"] != user_id]

    return {"message": "用户已删除"}

# 运行：uvicorn main:app --reload

使用 API

JavaScript Fetch

// GET 请求
async function getUsers() {
  try {
    const response = await fetch('http://localhost:3000/api/users');

    if (!response.ok) {
      throw new Error(HTTP 错误！状态：${response.status});
    }

    const data = await response.json();
    console.log(data);
    return data;
  } catch (error) {
    console.error('错误：', error);
  }
}

// POST 请求
async function createUser(userData) {
  try {
    const response = await fetch('http://localhost:3000/api/users', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json'
      },
      body: JSON.stringify(userData)
    });

    if (!response.ok) {
      throw new Error(HTTP 错误！状态：${response.status});
    }

    const data = await response.json();
    console.log('用户已创建：', data);
    return data;
  } catch (error) {
    console.error('错误：', error);
  }
}

// 使用
createUser({
  name: "王五",
  email: "wangwu@example.com"
});

// PATCH 请求
async function updateUserEmail(userId, email) {
  const response = await fetch(http://localhost:3000/api/users/${userId}, {
    method: 'PATCH',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ email })
  });

  return await response.json();
}

// DELETE 请求
async function deleteUser(userId) {
  const response = await fetch(http://localhost:3000/api/users/${userId}, {
    method: 'DELETE'
  });

  return await response.json();
}

Python requests

import requests

BASE_URL = "http://localhost:8000/api"

# GET 请求
def get_users():
    response = requests.get(f"{BASE_URL}/users")
    response.raise_for_status()
    return response.json()

# POST 请求
def create_user(name, email):
    data = {"name": name, "email": email}
    response = requests.post(f"{BASE_URL}/users", json=data)
    response.raise_for_status()
    return response.json()

# PATCH 请求
def update_user_email(user_id, email):
    data = {"email": email}
    response = requests.patch(f"{BASE_URL}/users/{user_id}", json=data)
    response.raise_for_status()
    return response.json()

# DELETE 请求
def delete_user(user_id):
    response = requests.delete(f"{BASE_URL}/users/{user_id}")
    response.raise_for_status()
    return response.json()

# 使用
try:
    users = get_users()
    print(f"用户：{users}")

    new_user = create_user("赵六", "zhaoliu@example.com")
    print(f"已创建：{new_user}")
except requests.exceptions.HTTPError as e:
    print(f"HTTP 错误：{e}")
except Exception as e:
    print(f"错误：{e}")

Axios (JavaScript)

import axios from 'axios';

const api = axios.create({
  baseURL: 'http://localhost:3000/api',
  headers: {
    'Content-Type': 'application/json'
  }
});

// GET
const users = await api.get('/users');
console.log(users.data);

// POST
const newUser = await api.post('/users', {
  name: "张三",
  email: "zhangsan@example.com"
});

// PATCH
await api.patch('/users/1', { email: "newemail@example.com" });

// DELETE
await api.delete('/users/1');

// 错误处理
try {
  const user = await api.get('/users/999');
} catch (error) {
  if (error.response) {
    console.log('错误：', error.response.data);
    console.log('状态：', error.response.status);
  }
}

分页

// Express.js
app.get('/api/users', (req, res) => {
  const page = parseInt(req.query.page) || 1;
  const limit = parseInt(req.query.limit) || 10;
  const startIndex = (page - 1)  limit;

  const endIndex = page  limit;

  const results = {
    success: true,
    data: users.slice(startIndex, endIndex),
    pagination: {
      currentPage: page,
      totalPages: Math.ceil(users.length / limit),
      totalItems: users.length,
      itemsPerPage: limit
    }
  };

  res.json(results);
});

// 使用：GET /api/users?page=2&limit=5

过滤和排序

app.get('/api/users', (req, res) => {
  let filtered = [...users];

  // 过滤
  if (req.query.name) {
    filtered = filtered.filter(u => 
      u.name.toLowerCase().includes(req.query.name.toLowerCase())
    );
  }

  // 排序
  if (req.query.sortBy) {
    const sortField = req.query.sortBy;
    const sortOrder = req.query.order === 'desc' ? -1 : 1;

    filtered.sort((a, b) => {
      if (a[sortField] < b[sortField]) return -1  sortOrder;

      if (a[sortField] > b[sortField]) return 1  sortOrder;
      return 0;
    });
  }

  res.json({ success: true, data: filtered });
});

// 使用：GET /api/users?name=张&sortBy=age&order=desc

身份验证

JWT (JSON Web Token)

import jwt from 'jsonwebtoken';

const SECRET_KEY = 'your-secret-key';

// 登录端点
app.post('/api/login', (req, res) => {
  const { email, password } = req.body;

  // 验证用户（示例）
  const user = users.find(u => u.email === email);

  if (!user || !verifyPassword(password)) {
    return res.status(401).json({
      success: false,
      error: '无效的凭据'
    });
  }

  // 生成 token
  const token = jwt.sign(
    { userId: user.id, email: user.email },
    SECRET_KEY,
    { expiresIn: '24h' }
  );

  res.json({
    success: true,
    token,
    user: { id: user.id, name: user.name, email: user.email }
  });
});

// 认证中间件
function authenticate(req, res, next) {
  const token = req.headers.authorization?.split(' ')[1];

  if (!token) {
    return res.status(401).json({
      success: false,
      error: '未提供 token'
    });
  }

  try {
    const decoded = jwt.verify(token, SECRET_KEY);
    req.user = decoded;
    next();
  } catch (error) {
    res.status(401).json({
      success: false,
      error: '无效的 token'
    });
  }
}

// 受保护的路由
app.get('/api/profile', authenticate, (req, res) => {
  const user = users.find(u => u.id === req.user.userId);
  res.json({ success: true, data: user });
});

最佳实践

1. 一致的响应格式

// 成功响应
{
  "success": true,
  "data": {...},
  "message": "操作成功"
}

// 错误响应
{
  "success": false,
  "error": "错误消息",
  "code": "ERROR_CODE"
}

2. 正确的 HTTP 状态码

• 200：成功
• 201：已创建
• 400：错误请求
• 401：未授权
• 403：禁止
• 404：未找到
• 500：服务器错误

3. 版本控制

// URL 版本控制
app.use('/api/v1/users', usersV1Router);
app.use('/api/v2/users', usersV2Router);

// 或标头版本控制
app.use((req, res, next) => {
  const version = req.headers['api-version'] || 'v1';
  req.apiVersion = version;
  next();
});

4. 速率限制

import rateLimit from 'express-rate-limit';

const limiter = rateLimit({
  windowMs: 15  60  1000, // 15 分钟
  max: 100, // 限制每个 IP 100 个请求
  message: '请求过多，请稍后再试'
});

app.use('/api/', limiter);

5. CORS

import cors from 'cors';

app.use(cors({
  origin: 'https://yourfrontend.com',
  methods: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE'],
  credentials: true
}));

测试 API

使用 curl

# GET
curl http://localhost:3000/api/users

# POST
curl -X POST http://localhost:3000/api/users \
  -H "Content-Type: application/json" \
  -d '{"name":"张三","email":"zhangsan@example.com"}'

# PATCH
curl -X PATCH http://localhost:3000/api/users/1 \
  -H "Content-Type: application/json" \
  -d '{"email":"newemail@example.com"}'

# DELETE
curl -X DELETE http://localhost:3000/api/users/1

使用 Postman

创建新请求

设置方法（GET、POST 等）

输入 URL

添加标头（Content-Type: application/json）

添加正文（JSON）

发送请求

结论

构建 JSON API 是现代 Web 开发的基本技能。通过遵循 REST 原则和最佳实践，您可以创建可扩展、可维护和安全的 API。

关键要点

使用正确的 HTTP 方法和状态码

保持一致的响应格式

实施适当的身份验证和授权

验证所有输入数据

处理错误优雅

记录您的 API

版本化您的端点

测试彻底

开始构建您的 JSON API，为您的应用程序提供强大的后端！