🚀 RESTful API 接口规范详解：构建高效、可扩展的 Web 服务（附 Python 源码）

一、 什么是 RESTful API？

二、 RESTful API 设计规范详解

1. URL 设计规范

# ❌ 错误的 URL 设计
GET /api/getUser?id=123           # 动词冗余
POST /api/updateUser/123         # 用 POST 做更新
GET /api/users/delete/123        # URL 包含动词

2. HTTP 方法语义

3. 状态码规范

4. 请求/响应设计

Authorization: Bearer <token>    # 认证
Content-Type: application/json   # 请求体格式
Accept: application/json         # 期望的响应格式

{
  "code": 200,            // 业务状态码（可选，HTTP状态码已足够）
  "message": "success",   // 人类可读的消息
  "data": {              // 响应的核心数据
    "id": 123,
    "name": "张三"
  },
  "meta": {              // 分页、时间戳等元信息
    "page": 1,
    "total": 100
  }
}

# 请求
GET /api/v1/users?page=2&page_size=20&sort=created_at&order=desc

# 响应
{
  "data": [...],
  "meta": {
    "page": 2,
    "page_size": 20,
    "total": 150,
    "total_pages": 8
  },
  "links": {
    "first": "/api/v1/users?page=1",
    "prev": "/api/v1/users?page=1",
    "next": "/api/v1/users?page=3",
    "last": "/api/v1/users?page=8"
  }
}

5. 版本管理

三、 Python 实战：Flask 实现完整 RESTful API

from flask import Flask, request, jsonify, abort
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
from datetime import datetime
import uuid
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///restapi.db'
app.config['JWT_SECRET_KEY'] = 'your-secret-key-here'  # 生产环境用强密钥
app.config['JSON_SORT_KEYS'] = False  # 保持 JSON 字段顺序

db = SQLAlchemy(app)
jwt = JWTManager(app)

# ==================== 数据模型 ====================
class User(db.Model):
    id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'created_at': self.created_at.isoformat() if self.created_at else None
        }

# ==================== 辅助函数 ====================
def paginate(query, page=1, per_page=20):
    """通用分页函数"""
    pagination = query.paginate(page=page, per_page=per_page, error_out=False)
    return {
        'data': [item.to_dict() for item in pagination.items],
        'meta': {
            'page': pagination.page,
            'per_page': pagination.per_page,
            'total': pagination.total,
            'total_pages': pagination.pages
        },
        'links': {
            'self': f"{request.base_url}?page={page}&per_page={per_page}",
            'next': f"{request.base_url}?page={pagination.next_num}&per_page={per_page}" if pagination.has_next else None,
            'prev': f"{request.base_url}?page={pagination.prev_num}&per_page={per_page}" if pagination.has_prev else None,
        }
    }

def validate_user_data(data, is_update=False):
    """请求体验证"""
    errors = {}
    
    if not is_update or 'username' in data:
        if not data.get('username') or len(data['username']) < 3:
            errors['username'] = '用户名至少3个字符'
    
    if not is_update or 'email' in data:
        if not data.get('email') or '@' not in data['email']:
            errors['email'] = '邮箱格式无效'
        elif User.query.filter_by(email=data['email']).first() and not is_update:
            errors['email'] = '邮箱已注册'
    
    if errors:
        abort(422, description={'errors': errors})

# ==================== 错误处理 ====================
@app.errorhandler(404)
def not_found(error):
    return jsonify({
        'code': 404,
        'message': 'Resource not found',
        'data': None
    }), 404

@app.errorhandler(422)
def unprocessable_entity(error):
    return jsonify({
        'code': 422,
        'message': 'Validation failed',
        'data': error.description
    }), 422

# ==================== 认证接口 ====================
@app.route('/api/v1/auth/login', methods=['POST'])
def login():
    """登录接口（非RESTful，但常见）"""
    data = request.get_json()
    user = User.query.filter_by(username=data.get('username')).first()
    
    if not user:
        abort(401, description='Invalid credentials')
    
    # 生产环境应验证密码哈希
    access_token = create_access_token(identity=user.id)
    return jsonify({
        'code': 200,
        'message': 'Login successful',
        'data': {'access_token': access_token, 'token_type': 'bearer'}
    }), 200

# ==================== 用户资源接口 ====================
@app.route('/api/v1/users', methods=['GET'])
@jwt_required()
def get_users():
    """获取用户列表（GET /users）"""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    
    # 过滤查询
    query = User.query
    if request.args.get('username'):
        query = query.filter(User.username.contains(request.args['username']))
    
    # 排序
    sort_by = request.args.get('sort_by', 'created_at')
    order = request.args.get('order', 'desc')
    if order == 'desc':
        query = query.order_by(db.desc(getattr(User, sort_by)))
    else:
        query = query.order_by(getattr(User, sort_by))
    
    return jsonify(paginate(query, page, per_page)), 200

@app.route('/api/v1/users', methods=['POST'])
def create_user():
    """创建用户（POST /users）"""
    data = request.get_json()
    validate_user_data(data, is_update=False)
    
    user = User(
        username=data['username'],
        email=data['email']
    )
    db.session.add(user)
    db.session.commit()
    
    # 201 Created，应在 Location 头包含新资源URL
    response = jsonify({
        'code': 201,
        'message': 'User created successfully',
        'data': user.to_dict()
    })
    response.headers['Location'] = f'/api/v1/users/{user.id}'
    return response, 201

@app.route('/api/v1/users/<user_id>', methods=['GET'])
@jwt_required()
def get_user(user_id):
    """获取单个用户（GET /users/{id}）"""
    user = User.query.get_or_404(user_id)
    return jsonify({
        'code': 200,
        'message': 'success',
        'data': user.to_dict()
    }), 200

@app.route('/api/v1/users/<user_id>', methods=['PUT'])
@jwt_required()
def update_user_put(user_id):
    """全量更新用户（PUT /users/{id}）"""
    user = User.query.get_or_404(user_id)
    data = request.get_json()
    
    # PUT 需要所有必要字段
    required_fields = ['username', 'email']
    if not all(field in data for field in required_fields):
        abort(400, description='PUT requires all fields: ' + ', '.join(required_fields))
    # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    validate_user_data(data, is_update=True)
    
    # 全量替换
    user.username = data['username']
    user.email = data['email']
    db.session.commit()
    
    return jsonify({
        'code': 200,
        'message': 'User updated successfully',
        'data': user.to_dict()
    }), 200

@app.route('/api/v1/users/<user_id>', methods=['PATCH'])
@jwt_required()
def update_user_patch(user_id):
    """部分更新用户（PATCH /users/{id}）"""
    user = User.query.get_or_404(user_id)
    data = request.get_json()
    
    validate_user_data(data, is_update=True)
    
    # 只更新提供的字段
    if 'username' in data:
        user.username = data['username']
    if 'email' in data:
        user.email = data['email']
    
    db.session.commit()
    return jsonify({
        'code': 200,
        'message': 'User updated successfully',
        'data': user.to_dict()
    }), 200

@app.route('/api/v1/users/<user_id>', methods=['DELETE'])
@jwt_required()
def delete_user(user_id):
    """删除用户（DELETE /users/{id}）"""
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    
    # 204 No Content，无响应体
    return '', 204

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True, port=5000)

四、 API 测试示例（cURL）

# 1. 创建用户
curl -X POST http://localhost:5000/api/v1/users \
  -H "Content-Type: application/json" \
  -d '{"username":"testuser","email":"test@example.com"}'

# 2. 登录获取 Token
curl -X POST http://localhost:5000/api/v1/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username":"testuser"}'

# 3. 获取用户列表（带分页和认证）
curl -X GET "http://localhost:5000/api/v1/users?page=1&per_page=10&sort_by=created_at&order=desc" \
  -H "Authorization: Bearer <your-token>"

# 4. 部分更新用户
curl -X PATCH http://localhost:5000/api/v1/users/<user_id> \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <your-token>" \
  -d '{"email":"new@example.com"}'

# 5. 删除用户
curl -X DELETE http://localhost:5000/api/v1/users/<user_id> \
  -H "Authorization: Bearer <your-token>"

五、 生产级最佳实践

# requirements.txt
Flask==2.3.3
Flask-RESTful==0.3.10
Flask-JWT-Extended==4.5.2
Flask-SQLAlchemy==3.0.5
marshmallow==3.20.0  # 数据验证
apispec==6.3.0       # OpenAPI 生成
flask-limiter==3.3.1 # 限流

💡 总结

1. 清晰的资源设计（URL 即文档）

2. 正确的 HTTP 语义（方法对应操作）

3. 精确的状态码（自解释的响应）

4. 一致的响应格式（便于客户端处理）

5. 完善的错误处理（友好的错误信息）