MongoDB Atlas共享集群下Flask-MongoEngine更新用户points属性失败排查

问题总结 #

使用MongoDB Atlas免费共享集群，新增用户功能正常，但increase_points函数存在两个问题：

• 无法通过referral_code查询到用户（但submit函数中可通过email正常查询）
• 函数内创建txt文件的代码未生效

问题原因分析 #

1. 用户查询与更新逻辑错误：
   • increase_points中误用了原生pymongo的db.user.update_one()方法，但这里的db是Flask-MongoEngine实例，并非原生MongoDB数据库对象，语法不匹配导致更新无效。
2. 文件路径问题：
   • 使用相对路径创建文件时，Flask运行的工作目录可能并非预期的项目根目录，导致文件生成在未知路径，看起来像是未生效。

修复方案 #

1. 修正用户更新逻辑 #

使用Flask-MongoEngine提供的模型方法进行查询与更新，两种可选方式：

• 方式一：修改用户对象属性后直接保存（推荐，符合ORM使用习惯）
• 方式二：使用模型的update_one方法批量更新

2. 修复文件创建路径 #

通过app.root_path获取项目根目录，拼接绝对路径创建文件，确保文件生成在预期位置。

修复后的完整代码 #

from flask import Flask, request, redirect, url_for, flash, render_template
from flask_mongoengine import MongoEngine
import os
import string
import random
from datetime import datetime

import logging

logging.basicConfig(level=logging.DEBUG)

app = Flask(__name__)

app.config['MONGODB_SETTINGS'] = {
    'db': 'kanzy-db',
    'host': 'mongodb+srv://hhuseyinsenoll:kanzydb11@cluster0.k0dfbzu.mongodb.net/kanzy-db?retryWrites=true&w=majority',
    'username': 'hhuseyinsenoll',
    'password': 'kanzydb11',
    'authentication_source': 'admin'  # Assuming admin is your auth source
}

app.config['SECRET_KEY'] = 'your_real_random_secret_key'  # 替换为真实随机密钥
# Initialize MongoDB
db = MongoEngine(app)

# Define the User Model
class User(db.Document):
    email = db.StringField(required=True, unique=True)
    referral_code = db.StringField(required=True, unique=True)
    points = db.IntField(default=0)
    date_created = db.DateTimeField(required=True)

def generate_referral_code():
    return ''.join(random.choices(string.ascii_letters + string.digits, k=8))

@app.route('/', methods=['GET'])
def home():
    return app.send_static_file('index.html')


@app.route('/success/<referral_code>', methods=['GET'])
def success(referral_code):
    return render_template('success.html', referral_code=referral_code)

    
def increase_points(referral_code, points_to_add):
    logging.debug(f"Attempting to increase points for referral_code: {referral_code}")
    user = User.objects(referral_code=referral_code).first()
    
    # 使用绝对路径创建文件
    readme1_path = os.path.join(app.root_path, 'readme1.txt')
    with open(readme1_path, 'w') as f:
            f.write('Create a new text file!')
    
    if user:
        new_points = user.points + points_to_add
      
        # 使用绝对路径创建文件
        readme2_path = os.path.join(app.root_path, 'readme2.txt')
        with open(readme2_path, 'w') as f:
            f.write('Create a new text file!')
        
        logging.debug(f"Original points: {user.points}, New points: {new_points}")
        
        # 方式一：修改属性后保存（推荐）
        user.points = new_points
        user.save()
        
        # 方式二：批量更新（适合无需加载完整对象的场景）
        # User.objects(referral_code=referral_code).update_one(points=new_points)
        
        logging.debug(f"Points updated successfully for user: {user.email}")
    else:
        logging.debug(f"No user found with referral_code: {referral_code}")


@app.route('/submit', methods=['POST'])
def submit():
    if request.method == 'POST':
        email = request.form['email']
        referral_code_used = request.form.get('referral_code')
        
        existing_user = User.objects(email=email).first()
        
        if existing_user:
            flash('Email already registered.')
            return redirect(url_for('home'))
        
        if referral_code_used:
            # 去除推荐码两端空格，避免输入失误导致查询失败
            referral_code_used = referral_code_used.strip()
            logging.debug(f"Referral code used during registration: {referral_code_used}")
            increase_points(referral_code_used, points_to_add=1)
        
        new_user = User(
            email=email,
            referral_code=generate_referral_code(),
            date_created=datetime.utcnow()
        )

        new_user.save()
        flash('Successfully registered!')
        return redirect(url_for('success', referral_code=new_user.referral_code))


@app.route('/users', methods=['GET'])
def users_list():
    users = User.objects()  # Retrieve all users from MongoDB
    users_data = "<h1>All Users</h1><ul>"
    for user in users:
        users_data += f"<li>{user.email} - Points: {user.points} - Referral Code: {user.referral_code}</li>"
    users_data += "</ul>"
    return users_data

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

额外优化建议 #

• 务必将SECRET_KEY替换为真实的随机字符串，提升应用安全性。
• 对用户输入的推荐码做去空格处理，避免因用户输入失误导致查询失败（代码中已添加该逻辑）。

内容的提问来源于stack exchange，提问作者rraven-v2