用Python锁定sqlite3数据库(重新要求澄清)

Question

几个星期前,我在SO上发布了关于如何在python中锁定sqlite3数据库的问题:

如何在Python中锁定sqlite3数据库？

但是,我不太相信答案是有效的.或者,也许我只是误解了答案.

这是我遇到的情况:

• 我有一个数据库"测试"
• 在数据库"test"中有一个表"book"
• 表"book"中有两列:"title"和"checked_out_by"

然后我有一个功能如下:

def checkout(title, user):
    con = get_connection_from_db()
    with con:
        checked_out_by = get_checked_out_by(title)
        if checked_out_by == '': # If NOT checked out:
            checkout(title, user)
            print user, "checked out", title
        elif checked_out_by == 'user':
            print user, "already got it"
        else:
            print user, "can't check it out because", checked_out_by, "has it!"

因此checkout()函数首先验证书籍是否未签出,如果是,则签出书籍.请注意,我正在使用推荐的"with con:"技巧来确保所有内容都是事务性的,快乐的和copacetic.

但是,我运行了一堆并发测试并发现了问题.具体来说,当我同时运行以下两个调用时:

checkout('foo', 'steve')
checkout('foo', 'tim')

输出表明它不能正常工作.我希望看到以下两个可能的输出中的一个:

steve checked out foo
tim can't check it out because steve has it!

要么:

tim checked out foo
steve can't check it out because tim has it!

但偶尔,我会得到这个输出:

tim checked out foo
steve checked out foo

我认为'with con:'技巧将确保我的数据库调用将捆绑在一起.有人可以向我解释我是否/如何弄错了？如果是这样,有没有办法使这项工作？

Answer 1

'con con'不是这里想要的.(或这个线程锁定垃圾)

要获得特定时期的独占访问权限(而不仅仅是在进行单个查询/交易时),您需要这样做;

con = sqlite3.connect()
con.isolation_level = 'EXCLUSIVE'
con.execute('BEGIN EXCLUSIVE')
#exclusive access starts here. Nothing else can r/w the db, do your magic here.
con.commit()
con.close()

希望这可以节省一些人从我刚刚经历的搜索/实验中获益!

记住它不是独占的,直到你开始独占,它将保持独占,直到你关闭(或运行提交,我认为).如果您不确定,可以随时使用python interpreter/CL sqlite3应用程序进行测试.

Answer 2

需要注意的一件重要事情是，当数据库被锁定时，意味着它不接受多个写入者。然而，它确实接受多个读者。

检查事务是否按预期工作的一种简单方法是将值写入数据库，然后在事务代码完成之前引发异常。如果未写入该值，则交易正常。否则，就会出现问题。

数据库事务是一种乐观的并发方法，也就是说，它们仅在即将提交时才会失败。既然您似乎正在寻找一种悲观的方法，也许您应该尝试使用threading.Lock：

import threading
db_lock = threading.Lock()

def checkout(title, user):
    with db_lock:
        con = get_connection_from_db()
        with con:
            checked_out_by = get_checked_out_by(title)
            if checked_out_by == '': # If NOT checked out:
                checkout(title, user)
                print user, "checked out", title
            elif checked_out_by == 'user':
                print user, "already got it"
            else:
                print user, "can't check it out because", checked_out_by, "has it!"