异步协程完成后更新SQLite数据库表遇锁问题咨询

核心问题 #

你遇到的sqlite3.OperationalError: database is locked错误，大概率是异步任务aall_answers_all_docs内部也在操作同一个SQLite数据库。SQLite默认是单写模式，当异步任务里的数据库连接未正确释放，或与主线程的更新操作同时触发时，就会引发锁冲突。另外你的代码中asyncio.run()本身是阻塞调用，会等待协程完成后才执行后续更新，所以问题不在等待逻辑，而是数据库连接的管理方式。

具体修复步骤 #

1. 修复SQL注入风险（优先处理） #

当前用字符串拼接SQL语句的方式存在严重安全隐患，必须改用参数化查询：

# 插入操作的参数化写法
insert_query = """INSERT INTO batch_jobs
                  (job_name, user_id, status, type) 
                  VALUES 
                  (?, ?, ?, ?)"""
conn.execute(insert_query, (job, user_id, 'Running', 'clause_extract'))

# 更新操作的参数化写法
update_query = """UPDATE batch_jobs 
                  SET status = ?,
                      file_location = ?
                  WHERE id = ?"""
conn.execute(update_query, ('Finished', f'./data/csv_files/{file_name}', row_id))

2. 确保异步任务内的数据库连接正确释放 #

检查aall_answers_all_docs函数，如果里面有数据库操作，必须用上下文管理器with自动管理连接，确保使用后立即释放：

async def aall_answers_all_docs(doc_list, template):
    # 用with管理数据库连接，自动处理关闭
    with sqlite3.connect('database.db') as db:
        conn = db.cursor()
        # 执行你的数据库操作逻辑
        conn.execute(...)
        db.commit()
    # 其他异步业务逻辑
    ...

3. 统一数据库连接管理（可选优化） #

如果多个模块都要操作数据库，可以封装工具函数简化连接管理：

def get_db_connection():
    conn = sqlite3.connect('database.db')
    conn.row_factory = sqlite3.Row  # 可选：让查询结果支持字段名访问
    return conn

# 使用示例
with get_db_connection() as db:
    cursor = db.cursor()
    cursor.execute(insert_query, params)
    db.commit()
    row_id = cursor.lastrowid

4. 确认异步任务的阻塞逻辑 #

你的代码中asyncio.run()本身就是阻塞调用，会等待协程完全执行完毕才会执行后续的更新语句，等待逻辑是正常的，不需要额外使用asyncio.gather（除非你需要同时运行多个协程）。

修复后的完整代码示例 #

# 插入任务运行状态
with sqlite3.connect('database.db') as db:
    conn = db.cursor()
    insert_query = """INSERT INTO batch_jobs
                      (job_name, user_id, status, type) 
                      VALUES 
                      (?, ?, ?, ?)"""
    conn.execute(insert_query, (job, user_id, 'Running', 'clause_extract'))
    db.commit()
    row_id = conn.lastrowid
    print(f"Record inserted successfully into batch_job table, row count: {conn.rowcount}")

# 执行异步任务（阻塞直到任务完成）
output = asyncio.run(aall_answers_all_docs(doc_list, template))

# 更新任务为完成状态
with sqlite3.connect('database.db') as db:
    conn = db.cursor()
    update_query = """UPDATE batch_jobs 
                      SET status = ?,
                          file_location = ?
                      WHERE id = ?"""
    conn.execute(update_query, ('Finished', f'./data/csv_files/{file_name}', row_id))
    db.commit()

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