Flask与SQLAlchemy：实现用户数据（积分）更新的实用指南

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

# 初始化Flask应用
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)

# 初始化SQLAlchemy
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, default=0) # 默认积分设为0

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

    def __repr__(self):
        return f"<User {self.name}>"

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

# 示例路由：显示点击器页面
@app.route("/clicker")
def clicker():
    # 假设这里会渲染一个包含点击按钮的HTML页面
    return render_template("clicker.html")

# 假设这是处理按钮点击的路由
@app.route('/buttonclick', methods=['POST'])
def buttonclick():
    print("Button pressed")

    # 从请求的JSON体中获取user_id
    # 实际应用中，你可能从session或JWT token中获取当前登录用户的ID
    user_id = request.json.get("user_id")

    if not user_id:
        return 'Error: user_id not provided', 400

    # 1. 查询用户
    # 使用filter_by根据_id字段查找用户
    user = users.query.filter_by(_id=user_id).first()

    if not user:
        return 'Error: User not found', 404

    # 2. 修改用户数据
    user.score += 1 # 将积分增加1

    # 3. 提交事务
    # 将会话中的更改保存到数据库
    db.session.commit()

    return 'Success', 200

# ... (之前的导入、app和db初始化、users模型定义) ...

@app.route('/buttonclick', methods=['POST'])
def buttonclick():
    print("Button pressed")

    # 假设user_id从请求的JSON体中获取
    # 在生产环境中，更推荐从会话或认证令牌中获取当前用户的ID
    user_id = request.json.get("user_id")

    if user_id is None:
        return 'Error: user_id is required in request body', 400

    # 查找用户并更新积分
    try:
        user = users.query.filter_by(_id=user_id).first()
        if user:
            user.score += 1
            db.session.commit()
            return 'Score updated successfully', 200
        else:
            return 'User not found', 404
    except Exception as e:
        db.session.rollback() # 发生错误时回滚事务
        return f'An error occurred: {str(e)}', 500

# ... (之前的导入、app和db初始化、users模型定义) ...

@app.route('/buttonclick_concurrent_safe', methods=['POST'])
def buttonclick_concurrent_safe():
    print("Concurrent safe button pressed")

    user_id = request.json.get("user_id")

    if user_id is None:
        return 'Error: user_id is required in request body', 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 'Score updated successfully (concurrent safe)', 200
        else:
            return 'User not found', 404
    except Exception as e:
        db.session.rollback() # 发生错误时回滚事务，释放锁
        return f'An error occurred: {str(e)}', 500