Python SQLite3插入后查询获过时值的问题排查

在Python的SQLite3中遇到异常行为：搭建的小型Flask服务包含addTask和deleteTask两个接口，数据库的tasks_order字段存储任务排序数组。
addTask接口会向tasks_order数组中添加随机生成的任务ID并更新数据库；deleteTask接口则从数据库获取tasks_order数组，移除指定任务ID后更新数据库。
数据库连接处理遵循Flask官方文档规范。当用Postman模拟20个用户并发调用两个接口时，初期请求正常，后续deleteTask接口报错：尝试移除不存在的任务ID。日志显示addTask已成功插入任务ID237，但后续deleteTask的新连接查询到的数组中无该ID，为何新连接无法读取数据库变更？

相关代码 #

addTask接口 #

@tasks_bp.route("/addTask", methods=["POST"])
def addTask():
    hash = ''.join(secrets.choice(string.ascii_letters + string.digits + string.punctuation) for _ in range(10))
    data = request.get_json()
    return tasksService.addTask(hash)

addTask服务逻辑 #

def addTask(hash):
    try:
        conn = db.getConn()
        new_task_id = random.randint(1, 999)
        tasks_order = dbm.getTasksOrderList()
        dbm.updateTasksOrderList(tasks_order + [new_task_id])
        conn.commit()
        tasks_order = dbm.getTasksOrderList()
        print(hash, "added", new_task_id, datetime.now().strftime("%M:%S.%f"))
        print(hash, tasks_order, datetime.now().strftime("%M:%S.%f"))
    except Exception as e:
        conn.rollback()
        raise e
    else:
        return {"id": new_task_id}

dbm模块 #

def getTasksOrderList():
    cur = db.getConn().cursor()
    row = cur.execute("SELECT order_list FROM tasks_order").fetchone()
    order_list_str = row["order_list"]
    return json.loads(order_list_str)


def updateTasksOrderList(order_list):
    order_list_str = json.dumps(order_list)
    cur = db.getConn().cursor()
    cur.execute("UPDATE tasks_order SET order_list = ?", (order_list_str,))

deleteTask接口 #

@tasks_bp.route("/deleteTask", methods=["POST"])
def deleteTask():
    try:
        hash = ''.join(secrets.choice(string.ascii_letters + string.digits + string.punctuation) for _ in range(10))
        data = request.get_json()
        return tasksService.deleteTask(hash, data["task_id"])
    except Exception as e:
        print(hash, "ERROR OCCURED", data["task_id"])

deleteTask服务逻辑 #

def deleteTask(hash, task_id):
    print(hash, "todelete", task_id, datetime.now().strftime("%M:%S.%f"))
    try:
        conn = db.getConn()
        tasks_order = dbm.getTasksOrderList()
        print(hash, tasks_order, task_id, datetime.now().strftime("%M:%S.%f"))
        tasks_order.remove(task_id)
        dbm.updateTasksOrderList(tasks_order)
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    else:
        return {}, 204

数据库连接处理 #

def getConn():
    db = getattr(g, '_database', None)
    if db is None:
        db = g._database = sqlite3.connect(DATABASE)
        db.row_factory = sqlite3.Row
    return db

@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

日志信息 #

?,PXYJ*pse added 237 02:51.805480
?,PXYJ*pse [54, 950, 86, 678, 417, 404, 237] 02:51.805480 # 此处查询数组包含237
t"qu5NI+bb todelete 237 02:51.903749
t"qu5NI+bb [54, 950, 86, 678, 417, 151] 237 02:51.905749 # 后续新连接查询数组无237
t"qu5NI+bb ERROR OCCURED 237

核心问题 #

1. SQLite默认隔离级别限制：SQLite默认用DEFERRED隔离级别，事务直到写操作才会获取锁。并发时多个连接可能读到旧数据（脏读），后续更新会覆盖之前的变更，导致数据丢失。
2. 事务快照隔离：SQLite的事务一旦开启，读操作只会看到事务启动时的数据库状态。即使其他连接提交了变更，当前事务也无法感知，除非重新开启事务或切换隔离级别。
3. 并发更新冲突：你的代码是读取整个数组、修改、再覆盖写入，这种“读-改-写”逻辑在并发场景下必然会出现丢失更新——比如A添加237后提交，B同时读取旧数组添加151并提交，直接覆盖掉A的变更。

解决方案 #

1. 调整事务隔离级别 #

将连接隔离级别改为READ COMMITTED，确保每次读操作都能获取最新的已提交数据：

def getConn():
    db = getattr(g, '_database', None)
    if db is None:
        db = g._database = sqlite3.connect(DATABASE)
        db.row_factory = sqlite3.Row
        # 设置隔离级别为READ COMMITTED
        db.isolation_level = 'READ COMMITTED'
    return db

2. 添加排他锁避免并发冲突 #

因为是对全局状态的修改，必须确保同一时间只有一个请求能修改数据。修改getTasksOrderList，读取时开启排他事务：

def getTasksOrderList():
    conn = db.getConn()
    cur = conn.cursor()
    # 开启排他事务，阻止其他写操作，确保读到最新数据
    cur.execute("BEGIN EXCLUSIVE")
    row = cur.execute("SELECT order_list FROM tasks_order").fetchone()
    order_list_str = row["order_list"]
    return json.loads(order_list_str)

此方式会降低并发性能，但能彻底避免冲突，适合小型服务。

3. 重构数据存储结构（推荐） #

不要将整个数组存在单个字段中，改用独立表存储每个任务的排序信息：

• 创建task_order表，包含task_id和position字段
• 新增任务时，插入记录并设置position为当前最大值+1
• 删除任务时，直接删除对应记录，并更新后续任务的position减1
  这种方式利用SQL的原子操作从根源解决并发冲突问题。

4. 规范事务处理流程 #

确保所有数据操作都复用同一个请求上下文的连接，避免在dbm模块中重复获取连接；同时保证所有代码路径都正确提交或回滚事务，避免事务未关闭导致的快照隔离问题。

内容的提问来源于stack exchange，提问作者user2401856