如何在Django中测试关闭数据库连接的方法？

Question

我有一个使用Django ORM的长期运行的Python进程.它看起来像这样:

import django
from django.db import connection

from my_app import models


def my_process():
    django.setup()
    while (True):
        do_stuff()
        connection.close()
        models.MyDjangoModel().save()

有时do_stuff需要很长时间,此时我遇到了一个错误,我的MySql连接超时,因为数据库服务器将连接作为空闲终止.添加connection.close()行强制django每次都获得一个新连接并修复该问题.(参见https://code.djangoproject.com/ticket/21597).

但是,我正在使用a测试此进程django.test.TestCase,并且调用connection.close导致这些测试失败,因为django的TestCase类在事务中包装测试,并且在该事务中关闭连接会导致事务中断并引发a django.db.transaction.TransactionManagementError.

在解决这个问题,我试过的尝试是设置CONN_MAX_AGE数据库参数并调用connection.close_if_unusable_or_obsolete代替,但交易也改变了连接的autocommit距离设置的默认值设置True到False这反过来又导致close_if_unusable_or_obsolete试试,反正(关闭连接https://开头的github .com/django/django/blob/master/django/db/backends/base/base.py#L497).

我想我也可以connection.close在测试中嘲笑所以它什么都不做,但这看起来有点像hacky.

测试需要关闭数据库连接的django方法的最佳方法是什么？

Answer 1

好吧,我不确定这是不是最好的答案,但由于这个问题到目前为止还没有回复,我将发布我最终用于后代的解决方案:

我创建了一个辅助函数,在关闭连接之前检查我们当前是否处于原子块中:

import django
from django.db import connection

from my_app import models


def close_connection():
    """Closes the connection if we are not in an atomic block.

    The connection should never be closed if we are in an atomic block, as
    happens when running tests as part of a django TestCase. Otherwise, closing
    the connection is important to avoid a connection time out after long actions.     
    Django does not automatically refresh a connection which has been closed 
    due to idleness (this normally happens in the request start/finish part 
    of a webapp's lifecycle, which this process does not have), so we must
    do it ourselves if the connection goes idle due to stuff taking a really 
    long time.
    """
    if not connection.in_atomic_block:
        connection.close()


def my_process():
    django.setup()
    while (True):
        do_stuff()
        close_connection()
        models.MyDjangoModel().save()

正如评论所述,close_connection防止connection.close被测试,所以我们不再破坏测试事务.