Django 测试“数据库表被锁定”

Question

在我的 Django 项目中，我有一个观点，即当用户发布 zip 文件时，它会立即回复，然后在线程的帮助下在后台处理数据。该视图在正常测试中工作正常，但是当我运行 Django 的测试时，它失败并出现 database table is locked 错误。目前，我正在使用默认的 SQLite 数据库，我知道如果我切换到另一个数据库，这个问题可能会得到解决，但我正在寻找当前设置的答案。为了简单起见，我修剪了代码。

看来问题出在DeviceReportModel 表中。但我不确定为什么TestDeviceReport 访问它。

Model.py:

class DeviceReportModel(models.Model):
    device_id = models.PositiveIntegerField(primary_key=True)
    ip = models.GenericIPAddressField()
    created_time = models.DateTimeField(default=timezone.now)
    report_file = models.FileField(upload_to="DeviceReport")
    device_datas = models.ManyToManyField(DeviceDataReportModel)

    def __str__(self):
        return str(self.id)

Serializers.py:

class DeviceReportSerializer(serializers.ModelSerializer):
    class Meta:
        model = DeviceReportModel
        fields = '__all__'
        read_only_fields = ('created_time', 'ip', 'device_datas')

views.py:

from django.utils import timezone
from django.core.files.base import ContentFile
from rest_framework.response import Response
from rest_framework import status, generics
import time
import threading
from queue import Queue

class DeviceReportHandler:
    ReportQueue = Queue()

    @staticmethod
    def save_datas(device_object, request_ip, b64datas):
        device_data_models = []
        # ...
        # process device_data_models
        # this will take some time
        time.sleep(10)
        return device_data_models

    @classmethod
    def Check(cls):
        while(True):
            if not cls.ReportQueue.empty():
                report = cls.ReportQueue.get()
                # ...
                report_model = DeviceReportModel(
                    device_id=report['device_object'], ip=report['request_ip'])
                # THIS LINE GIVES ERROR
                report_model.report_file.save(
                    "Report_{}.txt.gz".format(timezone.now()), ContentFile(report['report_data']))

                device_data_models = cls.save_datas(
                    report['device_object'], report['request_ip'], 'SomeData')
                report_model.device_datas.set(device_data_models)
                report_model.save()
                print("Report Handle Done")
            time.sleep(.1)

    @classmethod
    def run(cls):
        thr = threading.Thread(target=cls.Check)
        thr.daemon = True
        thr.start()


class DeviceReportView(generics.ListCreateAPIView):
    queryset = DeviceReportModel.objects.all()
    serializer_class = DeviceReportSerializer
    DeviceReportHandler.run()

    def post(self, request):
        # ...
        report = {
            'device_object': 1,
            'request_ip': '0.0.0.0',
            'report_data': b'Some report plain data',
        }
        # add request to ReportQueue
        DeviceReportHandler.ReportQueue.put(report)
        return Response("OK", status.HTTP_201_CREATED)

tests.py:

from rest_framework.test import APITestCase
import gzip
from io import BytesIO
import base64
import time

class TestDeviceReport(APITestCase):
    @classmethod
    def setUpTestData(cls):
        # add a new test device for other tests
        pass

    def generate_device_data(self):
        # generate fake device data
        return ""

    def test_Report(self):
        # generate device data
        device_data = ''
        for i in range(10):
            device_data += self.generate_device_data() + '\n'

        buf = BytesIO()
        compressed = gzip.GzipFile(fileobj=buf, mode="wb")
        compressed.write(device_data.encode())
        compressed.close()
        b64data = base64.b64encode(buf.getvalue()).decode()
        data = {
            "device_id": 1,
            "report_data": b64data
        }
        response = self.client.post(
            '/device/reports/', data=data, format='json')
        print(response.status_code, response.content)

    def tearDown(self):
        # put some sleep to check whether the data has been processed
        # see "Report Handle Done"
        time.sleep(10)

这是错误日志：

(myDjangoEnv) python manage.py test deviceApp.tests.tests.TestDeviceReport
Creating test database for alias 'default'...
System check identified no issues (0 silenced).
201 b'"OK"'
Exception in thread Thread-1:
Traceback (most recent call last):
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\backends\utils.py", line 84, in _execute       
    return self.cursor.execute(sql, params)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\backends\sqlite3\base.py", line 383, in execute
    return Database.Cursor.execute(self, query, params)
sqlite3.OperationalError: database table is locked

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\threading.py", line 917, in _bootstrap_inner
    self.run()
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "<project_path>\deviceApp\views.py", line 303, in Check
    "Report_{}.txt.gz".format(timezone.now()), ContentFile(report['report_data']))
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\models\fields\files.py", line 93, in save
    self.instance.save()
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\models\base.py", line 741, in save
    force_update=force_update, update_fields=update_fields)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\models\base.py", line 779, in save_base
    force_update, using, update_fields,
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\models\base.py", line 870, in _save_table
    result = self._do_insert(cls._base_manager, using, fields, update_pk, raw)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\models\base.py", line 908, in _do_insert
    using=using, raw=raw)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\models\manager.py", line 82, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\models\query.py", line 1186, in _insert
    return query.get_compiler(using=using).execute_sql(return_id)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\models\sql\compiler.py", line 1335, in execute_sql
    cursor.execute(sql, params)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\backends\utils.py", line 67, in execute
    return self._execute_with_wrappers(sql, params, many=False, executor=self._execute)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\backends\utils.py", line 76, in _execute_with_wrappers
    return executor(sql, params, many, context)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\backends\utils.py", line 84, in _execute
    return self.cursor.execute(sql, params)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\utils.py", line 89, in __exit__
    raise dj_exc_value.with_traceback(traceback) from exc_value
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\backends\utils.py", line 84, in _execute
    return self.cursor.execute(sql, params)
  File "C:\Users\Masoud\Anaconda3\envs\myDjangoEnv\lib\site-packages\django\db\backends\sqlite3\base.py", line 383, in execute
    return Database.Cursor.execute(self, query, params)
django.db.utils.OperationalError: database table is locked

.
----------------------------------------------------------------------
Ran 1 test in 10.023s

OK
Destroying test database for alias 'default'...

Answer 1

尝试使用django.test.TransactionTestCase 而不是TestCase

Answer 2

数据库被锁定错误

SQLite 旨在成为一个轻量级数据库，因此无法支持高级别的并发。 OperationalError: database is locked 错误表明您的应用程序遇到的并发性超出了 sqlite 在默认配置中可以处理的数量。这个错误意味着一个线程或进程在数据库连接上有一个排他锁，另一个线程超时等待锁被释放。

Python 的 SQLite 包装器有一个默认的超时值，它决定了第二个线程在超时之前允许在锁上等待多长时间并引发 OperationalError: database is locked 错误。

如果您遇到此错误，您可以通过以下方式解决：

切换到另一个数据库后端。在某个时刻，SQLite 对于现实世界的应用程序来说变得过于“精简”，而这些并发错误表明您已经达到了这一点。

重写代码以减少并发并确保数据库事务是短暂的。

通过设置超时数据库选项来增加默认超时值：

'OPTIONS': {
    # ...
    'timeout': 20,
    # ...
}

这将使 SQLite 在抛出“数据库已锁定”错误之前等待更长的时间；它不会真正解决它们。

https://docs.djangoproject.com/en/3.0/ref/databases/#database-is-locked-errorsoption