Python Flask HTTP 微服务开发实战

带路径和查询参数

实际开发中常需接收参数。Flask 支持路径参数和查询参数两种方式：

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/user/<username>', methods=['GET'])
def get_user(username):
    # 从查询字符串获取 age 参数，默认值为 18
    age = request.args.get('age', default=18, type=int)
    
    return jsonify({
        'username': username,
        'age': age,
        'message': f'用户 {username} 的信息',
        'status': 'success'
    })

访问 http://127.0.0.1:5000/api/user/zhangsan?age=25 可看到动态返回的数据。

[图片：处理带参数的 GET 接口]

POST 接口

POST 通常用于提交数据或创建资源。

处理 JSON 数据

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/user', methods=['POST'])
def create_user():
    data = request.get_json()
    
    if not data:
        return jsonify({'error': '没有提供数据'}), 400
    
    username = data.get('username')
    email = data.get('email')
    
    if not username or not email:
        return jsonify({'error': '用户名和邮箱是必填项'}), 400
    
    # 模拟数据库操作
    user_data = {
        'id': 1,
        'username': username,
        'email': email,
        'created_at': '2024-01-01'
    }
    
    return jsonify({
        'message': '用户创建成功',
        'user': user_data,
        'status': 'success'
    }), 201

建议使用 Postman 等工具测试，设置 Header 为 Content-Type: application/json。

[图片：POST 接口处理 JSON 数据]

处理表单请求

除了 JSON，表单提交也是常见场景，常用于文件上传：

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/form/user', methods=['POST'])
def create_user_form():
    content_type = request.headers.get('Content-Type', '').lower()
    is_form_data = 'multipart/form-data' in content_type
    
    username = request.form.get('username')
    email = request.form.get('email')
    
    if not username or not email:
        return jsonify({'error': '字段不能为空'}), 400
    
    uploaded_files = {}
    if is_form_data:
        for filename, file in request.files.items():
            if file and file.filename:
                # 保存文件逻辑
                pass
    
    return jsonify({
        'message': '用户创建成功',
        'status': 'success'
    }), 201

[图片：POST 接口处理表单请求]

数据库集成

现代 Web 应用离不开数据库。Flask 本身不包含数据库功能，但通过 SQLAlchemy 扩展可以轻松集成。

配置与模型定义

本示例使用 SQLite，无需单独安装服务器，适合开发和测试：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

flaskApp = Flask(__name__)
flaskApp.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
flaskApp.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

sqliteDb = SQLAlchemy(flaskApp)

class User(sqliteDb.Model):
    id = sqliteDb.Column(sqliteDb.Integer, primary_key=True)
    username = sqliteDb.Column(sqliteDb.String(80), unique=True, nullable=False)
    email = sqliteDb.Column(sqliteDb.String(120), unique=True, nullable=False)
    created_at = sqliteDb.Column(sqliteDb.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 flaskApp.app_context():
    sqliteDb.create_all()

CRUD 操作

创建用户

@flaskApp.route('/api/create/user', methods=['POST'])
def create_user():
    data = request.get_json()
    required_fields = ['username', 'email']
    
    for field in required_fields:
        if field not in data or not data[field].strip():
            return jsonify({'error': f'{field}是必填项'}), 400
    
    # 唯一性检查
    if User.query.filter_by(username=data['username']).first():
        return jsonify({'error': '用户名已存在'}), 409
    
    new_user = User(username=data['username'], email=data['email'])
    sqliteDb.session.add(new_user)
    sqliteDb.session.commit()
    
    return jsonify({'message': '用户创建成功', 'user': new_user.to_dict()}), 201

[图片：Post 接口创建用户]

获取所有用户

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

[图片：Get 接口返回所有用户]

获取单个用户

@flaskApp.route('/api/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get(user_id)
    if user is None:
        return jsonify({'error': '用户不存在'}), 404
    return jsonify(user.to_dict())

[图片：Get 接口返回单个用户]

搜索用户

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

[图片：Get 接口搜索用户]

更新用户信息

PUT 方法通常用于完全替换资源：

@flaskApp.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    user = User.query.get(user_id)
    if user is None:
        return jsonify({'error': '用户不存在'}), 404
    
    data = request.get_json()
    
    if 'username' in data and data['username']:
        existing = User.query.filter(
            User.username == data['username'],
            User.id != user_id
        ).first()
        if existing:
            return jsonify({'error': '用户名已被使用'}), 409
        user.username = data['username']
    
    if 'email' in data and data['email']:
        existing = User.query.filter(
            User.email == data['email'],
            User.id != user_id
        ).first()
        if existing:
            return jsonify({'error': '邮箱已被使用'}), 409
        user.email = data['email']
    
    sqliteDb.session.commit()
    return jsonify({'message': '用户更新成功', 'user': user.to_dict()})

[图片：Put 接口更新用户]

删除用户

@flaskApp.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get(user_id)
    if user is None:
        return jsonify({'error': '用户不存在'}), 404
    
    sqliteDb.session.delete(user)
    sqliteDb.session.commit()
    
    return jsonify({'message': '用户删除成功'}), 200

总结

通过以上步骤，我们完成了 Flask 微服务的基础搭建、接口实现及数据库集成。在实际生产中，记得将 debug 关闭，并引入 WSGI 服务器（如 Gunicorn）部署应用。注意异常处理和日志记录，确保服务的稳定性。