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

pgpifvop  于 2023-10-21  发布在  Go
关注(0)|答案(3)|浏览(214)

在我的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'...
django

3条答案

按热度按时间
mwg9r5ms

mwg9r5ms1#

数据库被锁定错误

SQLite是一个轻量级数据库，因此不能支持高级别的并发。操作错误：databaseislocked错误指示您的应用程序正在经历比sqlite在默认配置中可以处理的更多的并发。此错误意味着一个线程或进程在数据库连接上具有独占锁，而另一个线程在等待释放该锁时超时。
Python的SQLite Package 器有一个默认的超时值，它决定了第二个线程在锁超时并引发OperationalError之前可以等待多长时间：数据库被锁定错误。
如果你得到这个错误，你可以通过以下方式解决：
切换到另一个数据库后端。到了某个时候，SQLite对于现实世界的应用程序来说变得太“精简”了，而这些并发错误表明您已经达到了这一点。
重写代码以减少并发性并确保数据库事务是短暂的。
通过设置timeout数据库选项来增加默认超时值：

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

这将使SQLite在抛出“数据库被锁定”错误之前等待更长的时间;它并不能真正解决问题
https://docs.djangoproject.com/en/3.0/ref/databases/#database-is-locked-errorsoption

赞(0）分享  回复(0）举报  2023-10-21
k75qkfdt

k75qkfdt2#

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

赞(0）分享  回复(0）举报  2023-10-21
blpfk2vs

blpfk2vs3#

TestCase使用事务来 Package 单元测试，确保所有测试的数据库配置一致。
TransactionTestCase（名字很奇怪）的意思是“用事务进行测试的测试用例”。没有事务，所以其他线程/进程不必担心被锁定。当然，数据库状态不容易回滚，但Django使用“serialized_rollback”来序列化数据库并恢复数据库（通过JSON），这显然比事务慢。

赞(0）分享  回复(0）举报  2023-10-21
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com