Python Flask 微服务开发实战：接口设计与数据库集成

POST 请求处理

POST 常用于创建资源，支持 JSON 和表单两种主流格式。

处理 JSON 数据时，使用 request.get_json() 解析请求体：

@app.route('/api/user', methods=['POST'])
def create_user():
    data = request.get_json()
    if not data:
        return jsonify({'error': 'No data'}), 400
    
    username = data.get('username')
    email = data.get('email')
    
    if not username or not email:
        return jsonify({'error': 'Missing fields'}), 400
        
    # 模拟保存逻辑
    user_data = {'id': 1, 'username': username, 'email': email}
    return jsonify({'message': 'Created', 'user': user_data}), 201

对于表单提交（如文件上传），需检查 Content-Type 并读取 request.form 或 request.files：

@app.route('/api/form', methods=['POST'])
def handle_form():
    content_type = request.headers.get('Content-Type', '').lower()
    if 'multipart/form-data' in content_type:
        files = request.files
        # 处理文件保存逻辑
    else:
        form_data = request.form
        # 处理普通表单字段
    return jsonify({'status': 'ok'})

数据库集成

现代 Web 应用离不开持久化存储。Flask 通过 SQLAlchemy 扩展轻松对接各类数据库。本例以 SQLite 为例，无需额外服务器，适合快速开发。

配置与模型

初始化 SQLAlchemy 并定义 User 模型：

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    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.strftime('%Y-%m-%d %H:%M:%S')
        }

with app.app_context():
    db.create_all()

CRUD 操作

遵循 RESTful 规范，不同 HTTP 方法对应不同操作。

创建用户：验证唯一性后插入数据。

@app.route('/api/users', methods=['POST'])
def create_user_db():
    data = request.get_json()
    if User.query.filter_by(username=data['username']).first():
        return jsonify({'error': 'Username exists'}), 409
    
    new_user = User(username=data['username'], email=data['email'])
    db.session.add(new_user)
    db.session.commit()
    return jsonify({'user': new_user.to_dict()}), 201

查询列表：返回所有记录或分页结果。

@app.route('/api/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify({'users': [u.to_dict() for u in users], 'count': len(users)})

查询单个：通过路径参数定位。

@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_user_detail(user_id):
    user = User.query.get(user_id)
    if not user:
        return jsonify({'error': 'Not found'}), 404
    return jsonify(user.to_dict())

搜索：支持模糊匹配。

@app.route('/api/users/search', methods=['GET'])
def search_users():
    username = request.args.get('username', '')
    query = User.query
    if username:
        query = query.filter(User.username.contains(username))
    return jsonify({'users': [u.to_dict() for u in query.all()]})

更新与删除：PUT 用于完全替换，DELETE 移除资源。

@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return jsonify({'error': 'Not found'}), 404
    
    data = request.get_json()
    if 'username' in data:
        user.username = data['username']
    if 'email' in data:
        user.email = data['email']
        
    db.session.commit()
    return jsonify({'user': user.to_dict()})

@app.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get(user_id)
    if user:
        db.session.delete(user)
        db.session.commit()
    return jsonify({'message': 'Deleted'})

以上涵盖了 Flask 微服务开发的核心流程。实际生产中，还需考虑异常处理、日志记录及安全认证等细节。