Python封装函数后SQLite3创建数据库立即提示database is locked问题

问题描述 #

脚本封装成函数后，创建玩家专属SQLite数据库时刚创建就报错sqlite3.OperationalError: database is locked。调用自定义的is_open()函数检测返回True，即使等待5秒后执行建表语句依然触发错误，原代码如下：

import sqlite3 as sq, os, sys, re, psutil
from time import sleep
currentdir = os.path.dirname(os.path.realpath(__file__))
parentdir = os.path.dirname(currentdir)
sys.path.append(parentdir)
#
def create_db(player):
    player = re.sub(' ','%20',player)
    if not os.path.exists(os.path.join(currentdir,'Players',player)):
        os.mkdir(os.path.join(currentdir,'Players',player))
    dbcon = sq.connect(os.path.join(currentdir,'Players',player,f'{player}-API.sqlite'))
    dbcur = dbcon.cursor()
    def is_open(path):
        for proc in psutil.process_iter():
            try:
                files = proc.open_files()
                if files:
                    for _file in files:
                        if _file.path == path:
                            return True
            except psutil.NoSuchProcess as err:
                print(err)
        return False
    print(is_open(os.path.join(currentdir,'Players',player,f'{player}-API.sqlite')))
    try:
        dbcur.execute("""CREATE TABLE IF NOT EXISTS "activities" (
            "date"  TEXT,
            "details"   TEXT,
            "text"  TEXT
        , "datetime"    INTEGER)""")
    except sq.OperationalError:
        print(f'Database error, waiting')
        sleep(5)
        dbcur.execute("""CREATE TABLE IF NOT EXISTS "activities" (
            "date"  TEXT,
            "details"   TEXT,
            "text"  TEXT
        , "datetime"    INTEGER)""")
    dbcon.commit()
    dbcon.close()
#
player = input(f'input player name to create files for> ')
create_db(player)

问题根源 #

1. 自身进程占用数据库：代码中先调用sq.connect()打开了数据库，之后再用is_open()检测时，检测到的是当前脚本自己的连接，所以返回True——并非外部进程锁定数据库。
2. 嵌套函数逻辑冗余：is_open()嵌套在create_db内部，且检测时机错误，导致误判锁定来源。

修复方案 #

方案一：调整检测与连接顺序 #

先检测数据库是否被外部进程占用，确认无占用后再创建连接，避免自身连接干扰检测结果：

import sqlite3 as sq, os, sys, re, psutil
from time import sleep

currentdir = os.path.dirname(os.path.realpath(__file__))
parentdir = os.path.dirname(currentdir)
sys.path.append(parentdir)

# 把检测函数提取到外部，避免嵌套作用域问题
def is_db_locked(path):
    for proc in psutil.process_iter():
        try:
            for _file in proc.open_files():
                if _file.path == path:
                    return True
        except psutil.NoSuchProcess:
            continue
    return False

def create_db(player):
    player = re.sub(' ','%20',player)
    player_dir = os.path.join(currentdir, 'Players', player)
    db_path = os.path.join(player_dir, f'{player}-API.sqlite')
    
    # 创建玩家目录
    if not os.path.exists(player_dir):
        os.mkdir(player_dir)
    
    # 等待外部进程释放数据库
    while is_db_locked(db_path):
        print(f"数据库{db_path}被外部进程占用，等待中...")
        sleep(2)
    
    # 此时再创建连接
    dbcon = sq.connect(db_path)
    dbcur = dbcon.cursor()
    
    try:
        dbcur.execute("""CREATE TABLE IF NOT EXISTS "activities" (
            "date" TEXT,
            "details" TEXT,
            "text" TEXT,
            "datetime" INTEGER
        )""")
    except sq.OperationalError as e:
        print(f"建表失败：{e}")
    finally:
        # 确保提交并关闭连接
        dbcon.commit()
        dbcon.close()

player = input('输入要创建文件的玩家名称> ')
create_db(player)

方案二：利用SQLite内置超时机制 #

SQLite连接支持timeout参数，设置后遇到锁定会自动等待指定时长，无需手动检测，代码更简洁：

import sqlite3 as sq, os, sys, re

currentdir = os.path.dirname(os.path.realpath(__file__))
parentdir = os.path.dirname(currentdir)
sys.path.append(parentdir)

def create_db(player):
    player = re.sub(' ','%20',player)
    player_dir = os.path.join(currentdir, 'Players', player)
    db_path = os.path.join(player_dir, f'{player}-API.sqlite')
    
    if not os.path.exists(player_dir):
        os.mkdir(player_dir)
    
    # 设置超时时间为5秒，遇到锁定自动等待
    dbcon = sq.connect(db_path, timeout=5)
    dbcur = dbcon.cursor()
    
    try:
        dbcur.execute("""CREATE TABLE IF NOT EXISTS "activities" (
            "date" TEXT,
            "details" TEXT,
            "text" TEXT,
            "datetime" INTEGER
        )""")
    except sq.OperationalError as e:
        print(f"超时后仍无法建表：{e}")
    finally:
        dbcon.commit()
        dbcon.close()

player = input('输入要创建文件的玩家名称> ')
create_db(player)

注意事项 #

• CREATE TABLE IF NOT EXISTS是幂等操作，无需重复执行，只要确保连接正常即可。
• 操作完成后务必调用close()关闭连接，避免自身进程长期占用数据库文件。
• 嵌套函数容易引发作用域和逻辑混乱，建议将通用功能（如检测函数）提取到外部。

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