We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Django celery fixup doesn't respect Django settings for PostgreSQL connections
When using the Django-Celery fixup to run background tasks for a Django web service, the tasks in the background do not respect the settings in Django for PostgreSQL (possibly other) connections.  Every task will always create a new connection no matter the Django settings.  Although it is possible to bypass this with the environment variable CELERY_DB_REUSE_MAX, it is preferred for it to follow the settings given in Django.

## Checklist

- [ ] I have included the output of ``celery -A proj report`` in the issue.
      (if you are not able to do this, then at least specify the Celery
       version affected).
Celery 4.0.2 with potentially all versions of Django (tested on 1.10.3 and 1.11.2)
- [X] I have verified that the issue exists against the `master` branch of Celery.

This line causes this "issue":
https://github.com/celery/celery/blob/master/celery/fixups/django.py#L186

## Steps to reproduce
Note that these steps require some monitoring service to be used, we have New Relic.
Note also that we use Heroku for this app in question.
1) Have a web facing process with Django that connects to your PostgreSQL database for ORM purposes
2) Have a worker process that also connects to the PostgreSQL for ORM purposes
3) Have the DATABASES['default']['CONN_MAX_AGE'] setting set to anything that isn't 0 (easiest to see with `None` for persistent connections)
4) Make multiple requests to the web portion of Django to cause some ORM activity (easiest to see if it happens on every request)
5) Get multiple tasks to execute on the worker that will cause some ORM activity (easiest to see if it happens on every task)
6) Use your monitoring service (New Relic in our case) to view a breakdown of all of the requests and worker activity.  In New Relic you can check this using the transaction tracing; select the endpoint/task that made the db queries and check the breakdown.

## Expected behavior
psycopg2:connect would occur rarely with an average calls per transaction <<< 1

## Actual behavior
psycopg2:connect occurs very rarely with an average calls per transaction of <<< 1 for the web processes.
psycopg2:connect occurs every time with an average calls per transaction of 1 for the worker processes.

## Potential Resolution
With my limited knowledge of Celery's inner workings, it feels like a fairly simple fix that I could make on a PR myself, but I wanted some input before I spend the time setting that all up.
This fix seems to work when monkey patched into the `DjangoWorkerFixup` class.
``` Python

def _close_database(self):
    try:
        # Use Django's built in method of closing old connections.
        # This ensures that the database settings are respected.
        self._db.close_old_connections()
    except AttributeError:
        # Legacy functionality if we can't use the old connections for whatever reason.
        for conn in self._db.connections.all():
            try:
                conn.close()
            except self.interface_errors:
                pass
            except self.DatabaseError as exc:
                str_exc = str(exc)
                if 'closed' not in str_exc and 'not connected' not in str_exc:
                    raise

celery.fixups.django.DjangoWorkerFixup._close_database = _close_database
```
Django celery fixup doesn't respect Django settings for PostgreSQL connections
When using the Django-Celery fixup to run background tasks for a Django web service, the tasks in the background do not respect the settings in Django for PostgreSQL (possibly other) connections.  Every task will always create a new connection no matter the Django settings.  Although it is possible to bypass this with the environment variable CELERY_DB_REUSE_MAX, it is preferred for it to follow the settings given in Django.

## Checklist

- [ ] I have included the output of ``celery -A proj report`` in the issue.
      (if you are not able to do this, then at least specify the Celery
       version affected).
Celery 4.0.2 with potentially all versions of Django (tested on 1.10.3 and 1.11.2)
- [X] I have verified that the issue exists against the `master` branch of Celery.

This line causes this "issue":
https://github.com/celery/celery/blob/master/celery/fixups/django.py#L186

## Steps to reproduce
Note that these steps require some monitoring service to be used, we have New Relic.
Note also that we use Heroku for this app in question.
1) Have a web facing process with Django that connects to your PostgreSQL database for ORM purposes
2) Have a worker process that also connects to the PostgreSQL for ORM purposes
3) Have the DATABASES['default']['CONN_MAX_AGE'] setting set to anything that isn't 0 (easiest to see with `None` for persistent connections)
4) Make multiple requests to the web portion of Django to cause some ORM activity (easiest to see if it happens on every request)
5) Get multiple tasks to execute on the worker that will cause some ORM activity (easiest to see if it happens on every task)
6) Use your monitoring service (New Relic in our case) to view a breakdown of all of the requests and worker activity.  In New Relic you can check this using the transaction tracing; select the endpoint/task that made the db queries and check the breakdown.

## Expected behavior
psycopg2:connect would occur rarely with an average calls per transaction <<< 1

## Actual behavior
psycopg2:connect occurs very rarely with an average calls per transaction of <<< 1 for the web processes.
psycopg2:connect occurs every time with an average calls per transaction of 1 for the worker processes.

## Potential Resolution
With my limited knowledge of Celery's inner workings, it feels like a fairly simple fix that I could make on a PR myself, but I wanted some input before I spend the time setting that all up.
This fix seems to work when monkey patched into the `DjangoWorkerFixup` class.
``` Python

def _close_database(self):
    try:
        # Use Django's built in method of closing old connections.
        # This ensures that the database settings are respected.
        self._db.close_old_connections()
    except AttributeError:
        # Legacy functionality if we can't use the old connections for whatever reason.
        for conn in self._db.connections.all():
            try:
                conn.close()
            except self.interface_errors:
                pass
            except self.DatabaseError as exc:
                str_exc = str(exc)
                if 'closed' not in str_exc and 'not connected' not in str_exc:
                    raise

celery.fixups.django.DjangoWorkerFixup._close_database = _close_database
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `celery/fixups/django.py`
Content:
```
1 """Django-specific customization."""
2 from __future__ import absolute_import, unicode_literals
3 
4 import os
5 import sys
6 import warnings
7 
8 from kombu.utils.imports import symbol_by_name
9 from kombu.utils.objects import cached_property
10 
11 from datetime import datetime
12 from importlib import import_module
13 
14 from celery import _state
15 from celery import signals
16 from celery.exceptions import FixupWarning, ImproperlyConfigured
17 
18 __all__ = ['DjangoFixup', 'fixup']
19 
20 ERR_NOT_INSTALLED = """\
21 Environment variable DJANGO_SETTINGS_MODULE is defined
22 but Django isn't installed.  Won't apply Django fix-ups!
23 """
24 
25 
26 def _maybe_close_fd(fh):
27     try:
28         os.close(fh.fileno())
29     except (AttributeError, OSError, TypeError):
30         # TypeError added for celery#962
31         pass
32 
33 
34 def _verify_django_version(django):
35     if django.VERSION < (1, 8):
36         raise ImproperlyConfigured('Celery 4.x requires Django 1.8 or later.')
37 
38 
39 def fixup(app, env='DJANGO_SETTINGS_MODULE'):
40     """Install Django fixup if settings module environment is set."""
41     SETTINGS_MODULE = os.environ.get(env)
42     if SETTINGS_MODULE and 'django' not in app.loader_cls.lower():
43         try:
44             import django  # noqa
45         except ImportError:
46             warnings.warn(FixupWarning(ERR_NOT_INSTALLED))
47         else:
48             _verify_django_version(django)
49             return DjangoFixup(app).install()
50 
51 
52 class DjangoFixup(object):
53     """Fixup installed when using Django."""
54 
55     def __init__(self, app):
56         self.app = app
57         if _state.default_app is None:
58             self.app.set_default()
59         self._worker_fixup = None
60 
61     def install(self):
62         # Need to add project directory to path
63         sys.path.append(os.getcwd())
64 
65         self._settings = symbol_by_name('django.conf:settings')
66         self.app.loader.now = self.now
67 
68         signals.import_modules.connect(self.on_import_modules)
69         signals.worker_init.connect(self.on_worker_init)
70         return self
71 
72     @property
73     def worker_fixup(self):
74         if self._worker_fixup is None:
75             self._worker_fixup = DjangoWorkerFixup(self.app)
76         return self._worker_fixup
77 
78     @worker_fixup.setter
79     def worker_fixup(self, value):
80         self._worker_fixup = value
81 
82     def on_import_modules(self, **kwargs):
83         # call django.setup() before task modules are imported
84         self.worker_fixup.validate_models()
85 
86     def on_worker_init(self, **kwargs):
87         self.worker_fixup.install()
88 
89     def now(self, utc=False):
90         return datetime.utcnow() if utc else self._now()
91 
92     def autodiscover_tasks(self):
93         from django.apps import apps
94         return [config.name for config in apps.get_app_configs()]
95 
96     @cached_property
97     def _now(self):
98         return symbol_by_name('django.utils.timezone:now')
99 
100 
101 class DjangoWorkerFixup(object):
102     _db_recycles = 0
103 
104     def __init__(self, app):
105         self.app = app
106         self.db_reuse_max = self.app.conf.get('CELERY_DB_REUSE_MAX', None)
107         self._db = import_module('django.db')
108         self._cache = import_module('django.core.cache')
109         self._settings = symbol_by_name('django.conf:settings')
110 
111         self.interface_errors = (
112             symbol_by_name('django.db.utils.InterfaceError'),
113         )
114         self.DatabaseError = symbol_by_name('django.db:DatabaseError')
115 
116     def django_setup(self):
117         import django
118         django.setup()
119 
120     def validate_models(self):
121         from django.core.checks import run_checks
122         self.django_setup()
123         run_checks()
124 
125     def install(self):
126         signals.beat_embedded_init.connect(self.close_database)
127         signals.worker_ready.connect(self.on_worker_ready)
128         signals.task_prerun.connect(self.on_task_prerun)
129         signals.task_postrun.connect(self.on_task_postrun)
130         signals.worker_process_init.connect(self.on_worker_process_init)
131         self.close_database()
132         self.close_cache()
133         return self
134 
135     def on_worker_process_init(self, **kwargs):
136         # Child process must validate models again if on Windows,
137         # or if they were started using execv.
138         if os.environ.get('FORKED_BY_MULTIPROCESSING'):
139             self.validate_models()
140 
141         # close connections:
142         # the parent process may have established these,
143         # so need to close them.
144 
145         # calling db.close() on some DB connections will cause
146         # the inherited DB conn to also get broken in the parent
147         # process so we need to remove it without triggering any
148         # network IO that close() might cause.
149         for c in self._db.connections.all():
150             if c and c.connection:
151                 self._maybe_close_db_fd(c.connection)
152 
153         # use the _ version to avoid DB_REUSE preventing the conn.close() call
154         self._close_database()
155         self.close_cache()
156 
157     def _maybe_close_db_fd(self, fd):
158         try:
159             _maybe_close_fd(fd)
160         except self.interface_errors:
161             pass
162 
163     def on_task_prerun(self, sender, **kwargs):
164         """Called before every task."""
165         if not getattr(sender.request, 'is_eager', False):
166             self.close_database()
167 
168     def on_task_postrun(self, sender, **kwargs):
169         # See https://groups.google.com/group/django-users/
170         #            browse_thread/thread/78200863d0c07c6d/
171         if not getattr(sender.request, 'is_eager', False):
172             self.close_database()
173             self.close_cache()
174 
175     def close_database(self, **kwargs):
176         if not self.db_reuse_max:
177             return self._close_database()
178         if self._db_recycles >= self.db_reuse_max * 2:
179             self._db_recycles = 0
180             self._close_database()
181         self._db_recycles += 1
182 
183     def _close_database(self):
184         for conn in self._db.connections.all():
185             try:
186                 conn.close()
187             except self.interface_errors:
188                 pass
189             except self.DatabaseError as exc:
190                 str_exc = str(exc)
191                 if 'closed' not in str_exc and 'not connected' not in str_exc:
192                     raise
193 
194     def close_cache(self):
195         try:
196             self._cache.close_caches()
197         except (TypeError, AttributeError):
198             pass
199 
200     def on_worker_ready(self, **kwargs):
201         if self._settings.DEBUG:
202             warnings.warn('Using settings.DEBUG leads to a memory leak, never '
203                           'use this setting in production environments!')
204
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/celery/fixups/django.py b/celery/fixups/django.py
--- a/celery/fixups/django.py
+++ b/celery/fixups/django.py
@@ -183,7 +183,7 @@
     def _close_database(self):
         for conn in self._db.connections.all():
             try:
-                conn.close()
+                conn.close_if_unusable_or_obsolete()
             except self.interface_errors:
                 pass
             except self.DatabaseError as exc:

{"golden_diff": "diff --git a/celery/fixups/django.py b/celery/fixups/django.py\n--- a/celery/fixups/django.py\n+++ b/celery/fixups/django.py\n@@ -183,7 +183,7 @@\n     def _close_database(self):\n         for conn in self._db.connections.all():\n             try:\n-                conn.close()\n+                conn.close_if_unusable_or_obsolete()\n             except self.interface_errors:\n                 pass\n             except self.DatabaseError as exc:\n", "issue": "Django celery fixup doesn't respect Django settings for PostgreSQL connections\nWhen using the Django-Celery fixup to run background tasks for a Django web service, the tasks in the background do not respect the settings in Django for PostgreSQL (possibly other) connections.  Every task will always create a new connection no matter the Django settings.  Although it is possible to bypass this with the environment variable CELERY_DB_REUSE_MAX, it is preferred for it to follow the settings given in Django.\r\n\r\n## Checklist\r\n\r\n- [ ] I have included the output of ``celery -A proj report`` in the issue.\r\n      (if you are not able to do this, then at least specify the Celery\r\n       version affected).\r\nCelery 4.0.2 with potentially all versions of Django (tested on 1.10.3 and 1.11.2)\r\n- [X] I have verified that the issue exists against the `master` branch of Celery.\r\n\r\nThis line causes this \"issue\":\r\nhttps://github.com/celery/celery/blob/master/celery/fixups/django.py#L186\r\n\r\n## Steps to reproduce\r\nNote that these steps require some monitoring service to be used, we have New Relic.\r\nNote also that we use Heroku for this app in question.\r\n1) Have a web facing process with Django that connects to your PostgreSQL database for ORM purposes\r\n2) Have a worker process that also connects to the PostgreSQL for ORM purposes\r\n3) Have the DATABASES['default']['CONN_MAX_AGE'] setting set to anything that isn't 0 (easiest to see with `None` for persistent connections)\r\n4) Make multiple requests to the web portion of Django to cause some ORM activity (easiest to see if it happens on every request)\r\n5) Get multiple tasks to execute on the worker that will cause some ORM activity (easiest to see if it happens on every task)\r\n6) Use your monitoring service (New Relic in our case) to view a breakdown of all of the requests and worker activity.  In New Relic you can check this using the transaction tracing; select the endpoint/task that made the db queries and check the breakdown.\r\n\r\n## Expected behavior\r\npsycopg2:connect would occur rarely with an average calls per transaction <<< 1\r\n\r\n## Actual behavior\r\npsycopg2:connect occurs very rarely with an average calls per transaction of <<< 1 for the web processes.\r\npsycopg2:connect occurs every time with an average calls per transaction of 1 for the worker processes.\r\n\r\n## Potential Resolution\r\nWith my limited knowledge of Celery's inner workings, it feels like a fairly simple fix that I could make on a PR myself, but I wanted some input before I spend the time setting that all up.\r\nThis fix seems to work when monkey patched into the `DjangoWorkerFixup` class.\r\n``` Python\r\n\r\ndef _close_database(self):\r\n    try:\r\n        # Use Django's built in method of closing old connections.\r\n        # This ensures that the database settings are respected.\r\n        self._db.close_old_connections()\r\n    except AttributeError:\r\n        # Legacy functionality if we can't use the old connections for whatever reason.\r\n        for conn in self._db.connections.all():\r\n            try:\r\n                conn.close()\r\n            except self.interface_errors:\r\n                pass\r\n            except self.DatabaseError as exc:\r\n                str_exc = str(exc)\r\n                if 'closed' not in str_exc and 'not connected' not in str_exc:\r\n                    raise\r\n\r\ncelery.fixups.django.DjangoWorkerFixup._close_database = _close_database\r\n```\nDjango celery fixup doesn't respect Django settings for PostgreSQL connections\nWhen using the Django-Celery fixup to run background tasks for a Django web service, the tasks in the background do not respect the settings in Django for PostgreSQL (possibly other) connections.  Every task will always create a new connection no matter the Django settings.  Although it is possible to bypass this with the environment variable CELERY_DB_REUSE_MAX, it is preferred for it to follow the settings given in Django.\r\n\r\n## Checklist\r\n\r\n- [ ] I have included the output of ``celery -A proj report`` in the issue.\r\n      (if you are not able to do this, then at least specify the Celery\r\n       version affected).\r\nCelery 4.0.2 with potentially all versions of Django (tested on 1.10.3 and 1.11.2)\r\n- [X] I have verified that the issue exists against the `master` branch of Celery.\r\n\r\nThis line causes this \"issue\":\r\nhttps://github.com/celery/celery/blob/master/celery/fixups/django.py#L186\r\n\r\n## Steps to reproduce\r\nNote that these steps require some monitoring service to be used, we have New Relic.\r\nNote also that we use Heroku for this app in question.\r\n1) Have a web facing process with Django that connects to your PostgreSQL database for ORM purposes\r\n2) Have a worker process that also connects to the PostgreSQL for ORM purposes\r\n3) Have the DATABASES['default']['CONN_MAX_AGE'] setting set to anything that isn't 0 (easiest to see with `None` for persistent connections)\r\n4) Make multiple requests to the web portion of Django to cause some ORM activity (easiest to see if it happens on every request)\r\n5) Get multiple tasks to execute on the worker that will cause some ORM activity (easiest to see if it happens on every task)\r\n6) Use your monitoring service (New Relic in our case) to view a breakdown of all of the requests and worker activity.  In New Relic you can check this using the transaction tracing; select the endpoint/task that made the db queries and check the breakdown.\r\n\r\n## Expected behavior\r\npsycopg2:connect would occur rarely with an average calls per transaction <<< 1\r\n\r\n## Actual behavior\r\npsycopg2:connect occurs very rarely with an average calls per transaction of <<< 1 for the web processes.\r\npsycopg2:connect occurs every time with an average calls per transaction of 1 for the worker processes.\r\n\r\n## Potential Resolution\r\nWith my limited knowledge of Celery's inner workings, it feels like a fairly simple fix that I could make on a PR myself, but I wanted some input before I spend the time setting that all up.\r\nThis fix seems to work when monkey patched into the `DjangoWorkerFixup` class.\r\n``` Python\r\n\r\ndef _close_database(self):\r\n    try:\r\n        # Use Django's built in method of closing old connections.\r\n        # This ensures that the database settings are respected.\r\n        self._db.close_old_connections()\r\n    except AttributeError:\r\n        # Legacy functionality if we can't use the old connections for whatever reason.\r\n        for conn in self._db.connections.all():\r\n            try:\r\n                conn.close()\r\n            except self.interface_errors:\r\n                pass\r\n            except self.DatabaseError as exc:\r\n                str_exc = str(exc)\r\n                if 'closed' not in str_exc and 'not connected' not in str_exc:\r\n                    raise\r\n\r\ncelery.fixups.django.DjangoWorkerFixup._close_database = _close_database\r\n```\n", "before_files": [{"content": "\"\"\"Django-specific customization.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport os\nimport sys\nimport warnings\n\nfrom kombu.utils.imports import symbol_by_name\nfrom kombu.utils.objects import cached_property\n\nfrom datetime import datetime\nfrom importlib import import_module\n\nfrom celery import _state\nfrom celery import signals\nfrom celery.exceptions import FixupWarning, ImproperlyConfigured\n\n__all__ = ['DjangoFixup', 'fixup']\n\nERR_NOT_INSTALLED = \"\"\"\\\nEnvironment variable DJANGO_SETTINGS_MODULE is defined\nbut Django isn't installed.  Won't apply Django fix-ups!\n\"\"\"\n\n\ndef _maybe_close_fd(fh):\n    try:\n        os.close(fh.fileno())\n    except (AttributeError, OSError, TypeError):\n        # TypeError added for celery#962\n        pass\n\n\ndef _verify_django_version(django):\n    if django.VERSION < (1, 8):\n        raise ImproperlyConfigured('Celery 4.x requires Django 1.8 or later.')\n\n\ndef fixup(app, env='DJANGO_SETTINGS_MODULE'):\n    \"\"\"Install Django fixup if settings module environment is set.\"\"\"\n    SETTINGS_MODULE = os.environ.get(env)\n    if SETTINGS_MODULE and 'django' not in app.loader_cls.lower():\n        try:\n            import django  # noqa\n        except ImportError:\n            warnings.warn(FixupWarning(ERR_NOT_INSTALLED))\n        else:\n            _verify_django_version(django)\n            return DjangoFixup(app).install()\n\n\nclass DjangoFixup(object):\n    \"\"\"Fixup installed when using Django.\"\"\"\n\n    def __init__(self, app):\n        self.app = app\n        if _state.default_app is None:\n            self.app.set_default()\n        self._worker_fixup = None\n\n    def install(self):\n        # Need to add project directory to path\n        sys.path.append(os.getcwd())\n\n        self._settings = symbol_by_name('django.conf:settings')\n        self.app.loader.now = self.now\n\n        signals.import_modules.connect(self.on_import_modules)\n        signals.worker_init.connect(self.on_worker_init)\n        return self\n\n    @property\n    def worker_fixup(self):\n        if self._worker_fixup is None:\n            self._worker_fixup = DjangoWorkerFixup(self.app)\n        return self._worker_fixup\n\n    @worker_fixup.setter\n    def worker_fixup(self, value):\n        self._worker_fixup = value\n\n    def on_import_modules(self, **kwargs):\n        # call django.setup() before task modules are imported\n        self.worker_fixup.validate_models()\n\n    def on_worker_init(self, **kwargs):\n        self.worker_fixup.install()\n\n    def now(self, utc=False):\n        return datetime.utcnow() if utc else self._now()\n\n    def autodiscover_tasks(self):\n        from django.apps import apps\n        return [config.name for config in apps.get_app_configs()]\n\n    @cached_property\n    def _now(self):\n        return symbol_by_name('django.utils.timezone:now')\n\n\nclass DjangoWorkerFixup(object):\n    _db_recycles = 0\n\n    def __init__(self, app):\n        self.app = app\n        self.db_reuse_max = self.app.conf.get('CELERY_DB_REUSE_MAX', None)\n        self._db = import_module('django.db')\n        self._cache = import_module('django.core.cache')\n        self._settings = symbol_by_name('django.conf:settings')\n\n        self.interface_errors = (\n            symbol_by_name('django.db.utils.InterfaceError'),\n        )\n        self.DatabaseError = symbol_by_name('django.db:DatabaseError')\n\n    def django_setup(self):\n        import django\n        django.setup()\n\n    def validate_models(self):\n        from django.core.checks import run_checks\n        self.django_setup()\n        run_checks()\n\n    def install(self):\n        signals.beat_embedded_init.connect(self.close_database)\n        signals.worker_ready.connect(self.on_worker_ready)\n        signals.task_prerun.connect(self.on_task_prerun)\n        signals.task_postrun.connect(self.on_task_postrun)\n        signals.worker_process_init.connect(self.on_worker_process_init)\n        self.close_database()\n        self.close_cache()\n        return self\n\n    def on_worker_process_init(self, **kwargs):\n        # Child process must validate models again if on Windows,\n        # or if they were started using execv.\n        if os.environ.get('FORKED_BY_MULTIPROCESSING'):\n            self.validate_models()\n\n        # close connections:\n        # the parent process may have established these,\n        # so need to close them.\n\n        # calling db.close() on some DB connections will cause\n        # the inherited DB conn to also get broken in the parent\n        # process so we need to remove it without triggering any\n        # network IO that close() might cause.\n        for c in self._db.connections.all():\n            if c and c.connection:\n                self._maybe_close_db_fd(c.connection)\n\n        # use the _ version to avoid DB_REUSE preventing the conn.close() call\n        self._close_database()\n        self.close_cache()\n\n    def _maybe_close_db_fd(self, fd):\n        try:\n            _maybe_close_fd(fd)\n        except self.interface_errors:\n            pass\n\n    def on_task_prerun(self, sender, **kwargs):\n        \"\"\"Called before every task.\"\"\"\n        if not getattr(sender.request, 'is_eager', False):\n            self.close_database()\n\n    def on_task_postrun(self, sender, **kwargs):\n        # See https://groups.google.com/group/django-users/\n        #            browse_thread/thread/78200863d0c07c6d/\n        if not getattr(sender.request, 'is_eager', False):\n            self.close_database()\n            self.close_cache()\n\n    def close_database(self, **kwargs):\n        if not self.db_reuse_max:\n            return self._close_database()\n        if self._db_recycles >= self.db_reuse_max * 2:\n            self._db_recycles = 0\n            self._close_database()\n        self._db_recycles += 1\n\n    def _close_database(self):\n        for conn in self._db.connections.all():\n            try:\n                conn.close()\n            except self.interface_errors:\n                pass\n            except self.DatabaseError as exc:\n                str_exc = str(exc)\n                if 'closed' not in str_exc and 'not connected' not in str_exc:\n                    raise\n\n    def close_cache(self):\n        try:\n            self._cache.close_caches()\n        except (TypeError, AttributeError):\n            pass\n\n    def on_worker_ready(self, **kwargs):\n        if self._settings.DEBUG:\n            warnings.warn('Using settings.DEBUG leads to a memory leak, never '\n                          'use this setting in production environments!')\n", "path": "celery/fixups/django.py"}], "after_files": [{"content": "\"\"\"Django-specific customization.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport os\nimport sys\nimport warnings\n\nfrom kombu.utils.imports import symbol_by_name\nfrom kombu.utils.objects import cached_property\n\nfrom datetime import datetime\nfrom importlib import import_module\n\nfrom celery import _state\nfrom celery import signals\nfrom celery.exceptions import FixupWarning, ImproperlyConfigured\n\n__all__ = ['DjangoFixup', 'fixup']\n\nERR_NOT_INSTALLED = \"\"\"\\\nEnvironment variable DJANGO_SETTINGS_MODULE is defined\nbut Django isn't installed.  Won't apply Django fix-ups!\n\"\"\"\n\n\ndef _maybe_close_fd(fh):\n    try:\n        os.close(fh.fileno())\n    except (AttributeError, OSError, TypeError):\n        # TypeError added for celery#962\n        pass\n\n\ndef _verify_django_version(django):\n    if django.VERSION < (1, 8):\n        raise ImproperlyConfigured('Celery 4.x requires Django 1.8 or later.')\n\n\ndef fixup(app, env='DJANGO_SETTINGS_MODULE'):\n    \"\"\"Install Django fixup if settings module environment is set.\"\"\"\n    SETTINGS_MODULE = os.environ.get(env)\n    if SETTINGS_MODULE and 'django' not in app.loader_cls.lower():\n        try:\n            import django  # noqa\n        except ImportError:\n            warnings.warn(FixupWarning(ERR_NOT_INSTALLED))\n        else:\n            _verify_django_version(django)\n            return DjangoFixup(app).install()\n\n\nclass DjangoFixup(object):\n    \"\"\"Fixup installed when using Django.\"\"\"\n\n    def __init__(self, app):\n        self.app = app\n        if _state.default_app is None:\n            self.app.set_default()\n        self._worker_fixup = None\n\n    def install(self):\n        # Need to add project directory to path\n        sys.path.append(os.getcwd())\n\n        self._settings = symbol_by_name('django.conf:settings')\n        self.app.loader.now = self.now\n\n        signals.import_modules.connect(self.on_import_modules)\n        signals.worker_init.connect(self.on_worker_init)\n        return self\n\n    @property\n    def worker_fixup(self):\n        if self._worker_fixup is None:\n            self._worker_fixup = DjangoWorkerFixup(self.app)\n        return self._worker_fixup\n\n    @worker_fixup.setter\n    def worker_fixup(self, value):\n        self._worker_fixup = value\n\n    def on_import_modules(self, **kwargs):\n        # call django.setup() before task modules are imported\n        self.worker_fixup.validate_models()\n\n    def on_worker_init(self, **kwargs):\n        self.worker_fixup.install()\n\n    def now(self, utc=False):\n        return datetime.utcnow() if utc else self._now()\n\n    def autodiscover_tasks(self):\n        from django.apps import apps\n        return [config.name for config in apps.get_app_configs()]\n\n    @cached_property\n    def _now(self):\n        return symbol_by_name('django.utils.timezone:now')\n\n\nclass DjangoWorkerFixup(object):\n    _db_recycles = 0\n\n    def __init__(self, app):\n        self.app = app\n        self.db_reuse_max = self.app.conf.get('CELERY_DB_REUSE_MAX', None)\n        self._db = import_module('django.db')\n        self._cache = import_module('django.core.cache')\n        self._settings = symbol_by_name('django.conf:settings')\n\n        self.interface_errors = (\n            symbol_by_name('django.db.utils.InterfaceError'),\n        )\n        self.DatabaseError = symbol_by_name('django.db:DatabaseError')\n\n    def django_setup(self):\n        import django\n        django.setup()\n\n    def validate_models(self):\n        from django.core.checks import run_checks\n        self.django_setup()\n        run_checks()\n\n    def install(self):\n        signals.beat_embedded_init.connect(self.close_database)\n        signals.worker_ready.connect(self.on_worker_ready)\n        signals.task_prerun.connect(self.on_task_prerun)\n        signals.task_postrun.connect(self.on_task_postrun)\n        signals.worker_process_init.connect(self.on_worker_process_init)\n        self.close_database()\n        self.close_cache()\n        return self\n\n    def on_worker_process_init(self, **kwargs):\n        # Child process must validate models again if on Windows,\n        # or if they were started using execv.\n        if os.environ.get('FORKED_BY_MULTIPROCESSING'):\n            self.validate_models()\n\n        # close connections:\n        # the parent process may have established these,\n        # so need to close them.\n\n        # calling db.close() on some DB connections will cause\n        # the inherited DB conn to also get broken in the parent\n        # process so we need to remove it without triggering any\n        # network IO that close() might cause.\n        for c in self._db.connections.all():\n            if c and c.connection:\n                self._maybe_close_db_fd(c.connection)\n\n        # use the _ version to avoid DB_REUSE preventing the conn.close() call\n        self._close_database()\n        self.close_cache()\n\n    def _maybe_close_db_fd(self, fd):\n        try:\n            _maybe_close_fd(fd)\n        except self.interface_errors:\n            pass\n\n    def on_task_prerun(self, sender, **kwargs):\n        \"\"\"Called before every task.\"\"\"\n        if not getattr(sender.request, 'is_eager', False):\n            self.close_database()\n\n    def on_task_postrun(self, sender, **kwargs):\n        # See https://groups.google.com/group/django-users/\n        #            browse_thread/thread/78200863d0c07c6d/\n        if not getattr(sender.request, 'is_eager', False):\n            self.close_database()\n            self.close_cache()\n\n    def close_database(self, **kwargs):\n        if not self.db_reuse_max:\n            return self._close_database()\n        if self._db_recycles >= self.db_reuse_max * 2:\n            self._db_recycles = 0\n            self._close_database()\n        self._db_recycles += 1\n\n    def _close_database(self):\n        for conn in self._db.connections.all():\n            try:\n                conn.close_if_unusable_or_obsolete()\n            except self.interface_errors:\n                pass\n            except self.DatabaseError as exc:\n                str_exc = str(exc)\n                if 'closed' not in str_exc and 'not connected' not in str_exc:\n                    raise\n\n    def close_cache(self):\n        try:\n            self._cache.close_caches()\n        except (TypeError, AttributeError):\n            pass\n\n    def on_worker_ready(self, **kwargs):\n        if self._settings.DEBUG:\n            warnings.warn('Using settings.DEBUG leads to a memory leak, never '\n                          'use this setting in production environments!')\n", "path": "celery/fixups/django.py"}]}