Flask SQLAlchemy 数据库字段值更新实践指南-Python教程-PHP中文网

Flask SQLAlchemy 数据库字段值更新实践指南

碧海醫心

发布： 2025-12-01 10:05:19

原创

228人浏览过

Flask SQLAlchemy 数据库字段值更新实践指南

本教程旨在详细指导如何在 flask 应用中利用 sqlalchemy 高效更新数据库中的特定字段值。我们将以一个用户积分累加功能为例，涵盖从获取用户id、查询目标记录、实现并发安全的数据修改到最终提交变更的完整流程，并强调并发控制和错误处理的关键实践，帮助开发者构建健壮的数据更新逻辑。

1. Flask 与 SQLAlchemy 环境配置

在 Flask 项目中，首先需要配置 SQLAlchemy 数据库连接和模型。以下是一个基本的设置示例，其中定义了一个 users 模型，包含用户ID、姓名、邮箱和积分等字段。

from flask import Flask, redirect, url_for, render_template, request, session, flash
from datetime import timedelta
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.secret_key = "your_secret_key" # 生产环境中应使用更复杂的密钥
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.sqlite3' # 使用 SQLite 数据库
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False # 禁用事件追踪，节省资源
app.permanent_session_lifetime = timedelta(minutes=5) # 设置会话生命周期

db = SQLAlchemy(app)

# 定义用户模型
class users(db.Model):
    _id = db.Column("id", db.Integer, primary_key=True)
    name = db.Column(db.String(100))
    email = db.Column(db.String(100))
    score = db.Column(db.Integer) # 用户积分字段

    def __init__(self, name, email, score):
        self.name = name
        self.email = email
        self.score = score

# 在应用上下文启动时创建数据库表（如果不存在）
with app.app_context():
    db.create_all()

登录后复制

2. 实现数据更新逻辑

核心的数据更新逻辑发生在用户触发特定操作时，例如点击按钮。在此场景中，我们需要根据用户ID找到对应的用户记录，然后修改其 score 字段，并提交到数据库。

2.1 获取用户标识

在进行数据更新前，必须明确要更新哪条记录。通常，我们会通过以下方式获取用户ID：

从请求体中获取 (POST/PUT请求)：适用于前端通过 AJAX 发送数据的情况。
从 URL 路径参数中获取：例如 /user/<int:user_id>/update_score。
从会话 (Session) 中获取：当用户已登录时，其ID通常存储在服务器会话中。

本教程将以从请求体中获取 user_id 为例进行演示。

2.2 查询并锁定记录

为了确保数据一致性和防止并发冲突，在更新数据时，特别是涉及到累加操作，建议使用数据库的行级锁定机制。SQLAlchemy 提供了 with_for_update() 方法来实现这一点。

AI Humanize

使用AI改写工具，生成不可被AI检测的文本内容

154

查看详情

@app.route('/buttonclick', methods=['POST'])
def buttonclick():
    print("button pressed")
    # 假设前端通过 JSON 发送 user_id
    user_id = request.json.get("user_id")

    if not user_id:
        return {'message': 'User ID is required'}, 400

    try:
        # 查找用户，并使用 with_for_update() 进行行级锁定
        # 这可以防止在当前事务完成前，其他并发请求修改同一行数据
        user = users.query.filter_by(_id=user_id).with_for_update().first()

        if user:
            user.score += 1 # 积分加一
            db.session.commit() # 提交会话，将更改保存到数据库
            return {'message': 'Score updated successfully', 'new_score': user.score}, 200
        else:
            return {'message': 'User not found'}, 404
    except Exception as e:
        db.session.rollback() # 发生异常时回滚会话
        print(f"Error updating score: {e}")
        return {'message': 'An error occurred during score update'}, 500

登录后复制

代码解释：

user_id = request.json.get("user_id"): 从传入的 JSON 请求体中提取 user_id。
users.query.filter_by(_id=user_id): 根据 _id 字段查询用户。
.with_for_update(): 这是关键步骤。它在数据库层面为当前查询命中的行添加一个排他锁（通常是 SELECT ... FOR UPDATE 语句）。这意味着在当前事务提交或回滚之前，其他尝试修改或锁定这些行的事务将被阻塞，从而有效避免了“读-改-写”操作中的竞态条件（Race Condition）。
.first(): 获取查询结果的第一条记录。
user.score += 1: 直接修改 Python 对象上的属性值。SQLAlchemy 会跟踪这些更改。
db.session.commit(): 将所有在当前会话中对模型对象进行的更改持久化到数据库。
db.session.rollback(): 当发生异常时，回滚当前会话中的所有操作，确保数据库状态的一致性。

3. 前端交互示例 (HTML/JavaScript)

为了触发上述 buttonclick 路由，前端页面需要一个按钮，并使用 JavaScript 发送 POST 请求。

<!-- clicker.html 示例 -->
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Clicker Game</title>
</head>
<body>
    <h1>Welcome to the Clicker!</h1>
    <p>Your current score: <span id="currentScore">Loading...</span></p>
    <button id="clickButton">Click Me!</button>

    <script>
        // 假设用户ID已知，例如从后端模板渲染过来或者通过登录状态获取
        const currentUserId = 1; // 示例用户ID，实际应动态获取

        document.getElementById('clickButton').addEventListener('click', function() {
            fetch('/buttonclick', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json',
                },
                body: JSON.stringify({ user_id: currentUserId }),
            })
            .then(response => response.json())
            .then(data => {
                if (data.message === 'Score updated successfully') {
                    alert('Score increased! New score: ' + data.new_score);
                    document.getElementById('currentScore').textContent = data.new_score;
                } else {
                    alert('Error: ' + data.message);
                }
            })
            .catch((error) => {
                console.error('Error:', error);
                alert('Failed to update score due to network error.');
            });
        });

        // 页面加载时获取初始分数（可选）
        // fetch('/get_user_score/' + currentUserId) ...
    </script>
</body>
</html>

登录后复制

4. 注意事项与最佳实践

用户 ID 获取的安全性：在生产环境中，user_id 不应直接从客户端请求中无验证地获取。对于已登录用户，更安全的做法是从 Flask 的 session 对象中获取当前登录用户的 ID。
错误处理：始终为数据库操作添加 try...except...finally 块，并在异常发生时执行 db.session.rollback()，以确保数据库状态的原子性。
并发控制：with_for_update() 对于高并发场景下的计数器或库存管理等操作至关重要，它能有效避免数据不一致问题。但请注意，过度使用锁可能会导致性能瓶颈或死锁，应根据实际业务需求权衡。
事务管理：SQLAlchemy 的会话 (db.session) 默认在每个请求结束后自动关闭。理解事务的生命周期对于管理数据库操作至关重要。
数据验证：在更新数据之前，对传入的数据进行严格的验证（例如，确保 user_id 是整数，并且存在于数据库中）。
API 设计：考虑将 user_id 作为 URL 路径参数传递，如 @app.route('/user/<int:user_id>/buttonclick', methods=['POST'])，这样 API 结构更清晰。

总结

通过本教程，我们学习了如何在 Flask 应用中使用 SQLAlchemy 实现数据库字段的更新操作。关键步骤包括：正确配置 Flask 和 SQLAlchemy、根据用户标识查询目标记录、利用 with_for_update() 实现并发安全的更新、修改模型属性以及最终提交事务。同时，我们也强调了错误处理、并发控制和安全获取用户标识等最佳实践，这些对于构建健壮和可维护的 Web 应用至关重要。掌握这些技能将帮助您高效地管理 Flask 应用中的数据持久化层。

以上就是Flask SQLAlchemy 数据库字段值更新实践指南的详细内容，更多请关注php中文网其它相关文章！