How to test a method in Django that closes a database connection?

Question

I have a lengthy Python process that uses Django ORM. It looks something like this:

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()

Sometimes it do_stufftakes a very long time, and at this moment I encountered an error when my connection to MySql was disconnected because the database server killed the connection as idle. Adding a line connection.close()causes django to get a new connection every time and fixes this problem. (See https://code.djangoproject.com/ticket/21597 ).

, 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 , , .

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 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 , .