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.jsonname = user_data.get('name')if not name:return jsonify({"error": "缺少 'name' 参数"}), 400new_user = User(name=name)db.session.add(new_user)try:db.session.commit()return jsonify({"message": "新增成功"}), 201except 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"}), 400user = User.query.get(user_id)if not user:return jsonify({"error": "User not found"}), 404user.name = nametry:db.session.commit()except Exception as e:db.session.rollback()return jsonify({"error": str(e)}), 500return jsonify({"message": "User updated successfully"}), 200

完整源码下载

注意点

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