Flask 实现增改及分页查询的完整 Demo

简介

本文将通过一个 Flask Demo 来展示如何使用 Flask 框架实现 RESTful API，包括增加、修改用户信息以及分页查询功能。我们将使用 Flask 的扩展 SQLAlchemy 来处理数据库操作。

环境准备

首先，确保你已安装 Flask 和 Flask-SQLAlchemy。如果尚未安装，可以通过以下命令安装：

pip install Flask Flask-SQLAlchemy

Flask 应用结构

创建一个名为 app.py 的文件，并编写以下代码：

# 引入 Flask 和 SQLAlchemy 相关模块
from flask import Flask, request, jsonify, abort
from flask_sqlalchemy import SQLAlchemy
import os

# 创建 Flask 应用实例
app = Flask(__name__)

# 配置数据库 URI 和 JSON 编码
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///data.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config['JSON_AS_ASCII'] = False  # flask 2.2.5以下 允许返回非ASCII字符
app.json.ensure_ascii = False # flask 2.3.0以上 允许返回非ASCII字符

# 创建 SQLAlchemy 实例绑定到应用
db = SQLAlchemy(app)

# 定义 User 模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(20))

# 创建数据库表
db.create_all()

# 增加用户
@app.route("/user/add", methods=['POST'])
def add_user():
    # 省略具体实现...

# 分页查询用户
@app.route("/users", methods=['GET'])
def get_users():
    # 省略具体实现...

# 更新用户信息
@app.route("/user/<int:user_id>", methods=['POST'])
def update_user(user_id):
    # 省略具体实现...

# 运行 Flask 应用
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=3200, debug=True)

增加用户

在 add_user 函数中，我们处理 POST 请求来增加新用户：

@app.route("/user/add", methods=['POST'])
def add_user():
    user_data = request.json
    name = user_data.get('name')
    if not name:
        return jsonify({
   "error": "缺少 'name' 参数"}), 400
    new_user = User(name=name)
    db.session.add(new_user)
    try:
        db.session.commit()
        return jsonify({
   "message": "新增成功"}), 201
    except Exception as e:
        db.session.rollback()
        return jsonify({
   "error": str(e)}), 500

分页查询用户

get_users 函数实现了用户信息的分页查询：

@app.route("/users", methods=['GET'])
def get_users():
    page = int(request.args.get('page', 1))
    per_page = int(request.args.get('per_page', 10))
    users = User.query.paginate(page=page, per_page=per_page, error_out=False)
    result = {
   
        'page': page,
        'per_page': per_page,
        'total': users.total,
        'list': [{
   'id': user.id, 'username': user.name} for user in users.items]
    }
    return jsonify(result)

更新用户信息

update_user 函数处理 POST 请求来更新用户信息：

@app.route("/user/<int:user_id>", methods=['POST'])
def update_user(user_id):
    user_data = request.get_json()
    name = user_data.get('name')
    if not name:
        return jsonify({
   "error": "Missing 'name' parameter"}), 400
    user = User.query.get(user_id)
    if not user:
        return jsonify({
   "error": "User not found"}), 404
    user.name = name
    try:
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        return jsonify({
   "error": str(e)}), 500
    return jsonify({
   "message": "User updated successfully"}), 200

完整源码下载

注意点

1. 数据库配置：确保 SQLALCHEMY_DATABASE_URI 配置正确，它定义了数据库的位置和类型。
2. 错误处理：在每个函数中，我们添加了错误处理逻辑，确保在发生错误时返回合适的 HTTP 状态码和错误信息。
3. JSON 处理：通过 request.json 和 jsonify 处理 JSON 数据，注意检查请求数据的有效性。
4. 幂等性：虽然我们使用了 POST 请求来更新用户信息，但这违反了 RESTful API 的幂等性原则。在实际应用中，应考虑使用 PUT 或 PATCH 方法。
5. 安全性：在生产环境中，确保采取适当的安全措施，如验证用户身份、防止 SQL 注入等。
6. 编码问题：通过设置 JSON_AS_ASCII 或 app.json.ensure_ascii 为 False 解决中文编码问题。

通过这个 Demo，我们展示了 Flask 应用的基本结构和如何实现 RESTful API 的增改及分页查询功能。希望这个 Demo 对你在使用 Flask 进行 Web 开发时有所帮助。