1

我有一个在 Django 中失败的简单测试用例：

模型（app/models.py）：

from django.db import models
class M(models.Model):
    condition = models.CharField(max_length=80, db_index=True)

测试（app/tests.py）：

from django.test import TestCase
import threading
import time
from app.models import M

def insert():
    time.sleep(0.3)
    ua = M(condition='x')
    ua.save()

class DjangoRaceTest(TestCase):
    def test_parallel(self):
        insert()  # <--- works
        #threading.Thread(target=insert).start()  # <-- fails

        for i in range(10):
            count = M.objects.all()
            if count:
                return True
            time.sleep(0.1)

        assert count

上面的代码通过；但是如果我注释掉直接调用 insert 而取消注释线程调用，断言失败，我得到：

Creating test database for alias 'default'...
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/jacobsen/testdjango/app/tests.py", line 10, in insert
    ua.save()
  File "/home/jacobsen/dj14/lib/python2.6/site-packages/Django-1.4-py2.6.egg/django/db/models/base.py", line 463, in save
    self.save_base(using=using, force_insert=force_insert, force_update=force_update)
  File "/home/jacobsen/dj14/lib/python2.6/site-packages/Django-1.4-py2.6.egg/django/db/models/base.py", line 551, in save_base
    result = manager._insert([self], fields=fields, return_id=update_pk, using=using, raw=raw)
  File "/home/jacobsen/dj14/lib/python2.6/site-packages/Django-1.4-py2.6.egg/django/db/models/manager.py", line 203, in _insert
    return insert_query(self.model, objs, fields, **kwargs)
  File "/home/jacobsen/dj14/lib/python2.6/site-packages/Django-1.4-py2.6.egg/django/db/models/query.py", line 1576, in insert_query
    return query.get_compiler(using=using).execute_sql(return_id)
  File "/home/jacobsen/dj14/lib/python2.6/site-packages/Django-1.4-py2.6.egg/django/db/models/sql/compiler.py", line 910, in execute_sql
    cursor.execute(sql, params)
  File "/home/jacobsen/dj14/lib/python2.6/site-packages/Django-1.4-py2.6.egg/django/db/backends/sqlite3/base.py", line 337, in execute
    return Database.Cursor.execute(self, query, params)
DatabaseError: no such table: app_m

显然这里有一些 Django 魔法让我失望，或者......什么？我已经在 OS X、Ubuntu 10.04、Django 1.2 和 1.4 的各种组合上对此进行了测试。

4

1 回答 1

2

答案似乎是内存中的sqlite数据库不能跨线程共享。

这是来自 1.1 的错误报告，其中核心开发人员基于SQLAlchemy 缺乏对它的支持得出结论，这不是一个错误（在没有来自 sqlite 引用的其他直接信息的情况下）：

Pysqlite connections do not support being moved between threads, unless the check_same_thread Pysqlite flag is set to False. In addition, when using an in-memory SQLite database, the full database exists only within the scope of a single connection. It is reported that an in-memory database does not support being shared between threads regardless of the check_same_thread flag - which means that a multithreaded application cannot share data from a :memory: database across threads unless access to the connection is limited to a single worker thread which communicates through a queueing mechanism to concurrent threads.

于 2012-06-26T19:09:35.410 回答