Flask微服务实战：从零搭建RESTful API的完整指南

Flask 是一个轻量级的 Python Web 框架，核心思路是”微内核 + 可扩展”。所谓”微”并不代表功能简陋，而是指框架本身只提供最基础的 Web 服务能力，其余功能都通过扩展生态按需引入。这种设计让开发者可以从几行代码的简单应用起步，逐步演进到复杂的企业级系统。

依赖安装与环境验证

开发前确保已安装 Python 3.7 或更高版本。通过 pip 安装 Flask：

pip install flask
pip3 install flask

安装完成后，用一段最小化代码验证 Flask 是否正常工作：

from flask import Flask

app = Flask(__name__)

@app.route('/', methods=['GET'])
def hello():
    """
    根路径 GET 请求处理函数。
    访问网站根路径时返回安装成功提示。
    """
    return "Flask安装成功！"

if __name__ == '__main__':
    app.run(debug=True)

运行后访问 http://127.0.0.1:5000，看到”Flask安装成功！”即说明环境就绪。

基础 HTTP 接口实现

环境就绪后，接下来实现最常用的 GET 和 POST 接口。

GET 接口

GET 请求用于从服务器获取数据。Flask 的路由装饰器语法可以直观地映射不同请求路径。

不带参数的 GET 接口

最基础的路由定义——不接收参数，直接返回固定 JSON：

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/hello', methods=['GET'])
def hello_world():
    """
    最简单的 GET 接口示例：不接收参数，返回固定欢迎消息。
    常用于接口连通性测试。
    """
    return jsonify({
        'message': 'Hello, World!',
        'method': 'GET',
        'status': 'success'
    })

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

使用示例：浏览器或工具访问 http://127.0.0.1:5000/api/hello 即可看到返回的 JSON。

带路径参数和查询参数的 GET 接口

实际开发中，GET 请求通常需要传参。Flask 支持路径参数（嵌入在 URL 中）和查询参数（?key=value 形式）：

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/user/<username>', methods=['GET'])
def get_user(username):
    """
    1. 路径参数：username 直接嵌入 URL 路径
    2. 查询参数：age 通过 URL 问号后的查询字符串传入
    """
    age = request.args.get('age', default=18, type=int)

    return jsonify({
        'username': username,
        'age': age,
        'message': f'用户 {username} 的信息',
        'method': 'GET',
        'status': 'success'
    })

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

使用示例：访问 http://127.0.0.1:5000/api/user/zhangsan?age=25，其中 zhangsan 是路径参数，age=25 是查询参数。

POST 接口

POST 请求用于向服务器提交数据，通常对应”创建新资源”。

处理 JSON 格式的 POST 请求

接收 JSON 请求体并做参数校验：

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/user', methods=['POST'])
def create_user():
    """
    POST 请求特点：
    1. 参数在请求体中，不会出现在 URL 里
    2. 通常用于创建或修改资源
    3. 支持 JSON、表单、文件等多种数据格式
    """
    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,
        'method': 'POST',
        'status': 'success'
    }), 201

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

Postman 测试配置：

POST http://127.0.0.1:5000/api/user
Headers：Content-Type: application/json
Body：{"username": "李四", "email": "lisi@example.com"}

处理表单提交的 POST 请求

除了 JSON，表单（form-data / x-www-form-urlencoded）也是 POST 的常见格式，适用于 HTML 表单和文件上传场景：

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/form/user', methods=['POST'])
def create_user_form():
    """通用表单处理接口，同时支持 urlencoded 和 multipart 两种格式"""

    content_type = request.headers.get('Content-Type', '').lower()

    is_form_urlencoded = 'application/x-www-form-urlencoded' in content_type
    is_form_data = 'multipart/form-data' in content_type

    if not (is_form_urlencoded or is_form_data):
        return jsonify({
            'error': '不支持的 Content-Type',
            'supported_types': [
                'application/x-www-form-urlencoded',
                'multipart/form-data'
            ]
        }), 415

    """
    POST 表单请求特点：
    1. 数据通过表单字段（form data）传递，不是 JSON
    2. Content-Type 通常为 application/x-www-form-urlencoded 或 multipart/form-data
    3. 适合 HTML 表单提交、文件上传等场景
    4. 数据格式为 key1=value1&key2=value2
    """

    username = request.form.get('username')
    email = request.form.get('email')

    if not username or username.strip() == '':
        return jsonify({'error': '用户名是必填项'}), 400
    if not email or email.strip() == '':
        return jsonify({'error': '邮箱是必填项'}), 400

    uploaded_files = {}
    if is_form_data:
        for filename, file in request.files.items():
            if file and file.filename:
                file.save(f'uploads/{file.filename}')
                uploaded_files[filename] = {
                    'filename': file.filename,
                    'content_type': file.content_type,
                    'size': len(file.read())
                }
                file.seek(0)

    user_data = {
        'id': 1,
        'username': username.strip(),
        'email': email.strip(),
        'created_at': '2024-01-01',
        'format': 'form-data' if is_form_data else 'x-www-form-urlencoded',
        'uploaded_files': uploaded_files
    }

    return jsonify({
        'message': '用户创建成功',
        'user': user_data,
        'method': 'POST',
        'status': 'success'
    }), 201

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

数据库集成

现代 Web 应用几乎都离不开数据库。Flask 通过扩展机制可以灵活接入多种数据库，本文以 SQLite 为例——它无需独立服务器，数据存于单文件，适合快速原型和本地开发。

配置与模型定义

from flask import Flask, jsonify, request
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()

创建用户（Create）

@flaskApp.route('/api/create/user', methods=['POST'])
def create_user():
    """
    创建新用户。
    RESTful 风格：POST /api/create/user 创建新资源。
    请求体需包含 JSON 格式的用户数据。
    """
    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

    if User.query.filter_by(email=data['email']).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

if __name__ == '__main__':
    flaskApp.run(debug=True, port=5000)

查询所有用户（Read）

@flaskApp.route('/api/users', methods=['GET'])
def get_all_users():
    """
    获取全部用户信息。
    RESTful 风格：GET /api/users 获取资源列表。
    """
    users = User.query.all()
    return jsonify({
        'users': [user.to_dict() for user in users],
        'count': len(users)
    })

if __name__ == '__main__':
    flaskApp.run(debug=True, port=5000)

查询单个用户（Read by ID）

RESTful 设计通过 URL 路径参数定位资源，语义清晰、便于缓存：

@flaskApp.route('/api/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    """
    获取单个用户的详细信息。
    RESTful 风格：GET /api/users/{id} 获取指定资源。
    user_id 从 URL 路径中获取。
    """
    user = User.query.get(user_id)

    if user is None:
        return jsonify({'error': '用户不存在'}), 404

    return jsonify(user.to_dict())

if __name__ == '__main__':
    flaskApp.run(debug=True, port=5000)

搜索用户（模糊查询）

通过查询参数过滤用户列表，支持模糊匹配：

@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
        }
    })

if __name__ == '__main__':
    flaskApp.run(debug=True, port=5000)

更新用户信息（Update）

PUT 语义更新客户端提供的字段，更新前检查唯一性约束：

@flaskApp.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    """
    更新用户信息。
    RESTful 风格：PUT /api/users/{id} 更新指定资源。
    PUT 通常用于整体替换，PATCH 用于部分更新。
    """
    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 = User.query.filter(
            User.username == data['username'],
            User.id != user_id
        ).first()
        if existing_user:
            return jsonify({'error': '用户名已被使用'}), 409
        user.username = data['username']

    if 'email' in data and data['email']:
        existing_user = User.query.filter(
            User.email == data['email'],
            User.id != user_id
        ).first()
        if existing_user:
            return jsonify({'error': '邮箱已被使用'}), 409
        user.email = data['email']

    sqliteDb.session.commit()

    return jsonify({
        'message': '用户更新成功',
        'user': user.to_dict()
    })

if __name__ == '__main__':
    flaskApp.run(debug=True, port=5000)

删除用户（Delete）

删除操作不可恢复，需要先确认资源存在：

@flaskApp.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    """
    删除用户。
    RESTful 风格：DELETE /api/users/{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

if __name__ == '__main__':
    flaskApp.run(debug=True, port=5000)

小结

从依赖安装到 CRUD 全套接口，Flask 用一套简洁的装饰器语法完成了 Web 微服务的完整搭建。代码中所有数据库操作都通过 SQLAlchemy ORM 完成，后续切换到 MySQL 或 PostgreSQL 时，通常只需修改连接字符串和少量配置，业务逻辑无需重写。这套基础骨架可以作为大多数 Python Web 项目的起点。