We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Database backend race condition during table creation
<!--
Please fill this template entirely and do not erase parts of it.
We reserve the right to close without a response
bug reports which are incomplete.
-->
# Checklist
<!--
To check an item on the list replace [ ] with [x].
-->
- [x] I have verified that the issue exists against the `master` branch of Celery.
- [ ] This has already been asked to the [discussion group](https://groups.google.com/forum/#!forum/celery-users) first.
- [x] I have read the relevant section in the
  [contribution guide](http://docs.celeryproject.org/en/latest/contributing.html#other-bugs)
  on reporting bugs.
- [x] I have checked the [issues list](https://github.com/celery/celery/issues?q=is%3Aissue+label%3A%22Issue+Type%3A+Bug+Report%22+-label%3A%22Category%3A+Documentation%22)
  for similar or identical bug reports.
- [x] I have checked the [pull requests list](https://github.com/celery/celery/pulls?q=is%3Apr+label%3A%22PR+Type%3A+Bugfix%22+-label%3A%22Category%3A+Documentation%22)
  for existing proposed fixes.
- [x] I have checked the [commit log](https://github.com/celery/celery/commits/master)
  to find out if the bug was already fixed in the master branch.
- [x] I have included all related issues and possible duplicate issues
  in this issue (If there are none, check this box anyway).

## Mandatory Debugging Information

- [x] 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).
- [x] I have verified that the issue exists against the `master` branch of Celery.
- [ ] I have included the contents of ``pip freeze`` in the issue. **N/A**
- [ ] I have included all the versions of all the external dependencies required
  to reproduce this bug. **N/A**

## Optional Debugging Information
<!--
Try some of the below if you think they are relevant.
It will help us figure out the scope of the bug and how many users it affects.
-->
- [ ] I have tried reproducing the issue on more than one Python version
  and/or implementation.
- [ ] I have tried reproducing the issue on more than one message broker and/or
  result backend.
- [ ] I have tried reproducing the issue on more than one version of the message
  broker and/or result backend.
- [ ] I have tried reproducing the issue on more than one operating system.
- [ ] I have tried reproducing the issue on more than one workers pool.
- [ ] I have tried reproducing the issue with autoscaling, retries,
  ETA/Countdown & rate limits disabled.
- [ ] I have tried reproducing the issue after downgrading
  and/or upgrading Celery and its dependencies.

## Related Issues and Possible Duplicates
<!--
Please make sure to search and mention any related issues
or possible duplicates to this issue as requested by the checklist above.

This may or may not include issues in other repositories that the Celery project
maintains or other repositories that are dependencies of Celery.

If you don't know how to mention issues, please refer to Github's documentation
on the subject: https://help.github.com/en/articles/autolinked-references-and-urls#issues-and-pull-requests
-->

#### Related Issues

- https://github.com/celery/celery/issues/4653

#### Possible Duplicates

- None

## Environment & Settings
<!-- Include the contents of celery --version below -->
**Celery version**: 4.4.7 (cliffs)
<!-- Include the output of celery -A proj report below -->
<details>
<summary><b><code>celery report</code> Output:</b></summary>
<p>

```
software -> celery:4.4.7 (cliffs) kombu:4.6.11 py:3.8.2
            billiard:3.6.3.0 py-amqp:2.6.1
platform -> system:Darwin arch:64bit
            kernel version:16.7.0 imp:CPython
loader   -> celery.loaders.app.AppLoader
settings -> transport:amqp results:db+postgresql://<redacted>

include: ['redacted']
accept_content: ['redacted-custom']
database_table_names: {
 'group': 'celery_group', 'task': 'celery_task'}
result_serializer: 'redacted-custom'
task_serializer: 'redacted-custom'
task_track_started: True
broker_url: 'amqp://<redacted>'
result_backend: 'db+postgresql://<redacted>'
```

</p>
</details>

# Steps to Reproduce

When celery uses a database result backend, the following line can be called multiple times from different processes:

https://github.com/celery/celery/blob/9a6c2923e859b6993227605610255bd632c1ae68/celery/backends/database/session.py#L56

This is a race condition because SQLAlchemy first checks if the tables/sequences exist and then tries to create them. It causes errors like this (at least on PostgreSQL):

```
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/redacted.py", line 168, in _redacted
    result = async_result.get()
  File "/usr/local/lib/python3.7/site-packages/celery/result.py", line 226, in get
    self.maybe_throw(callback=callback)
  File "/usr/local/lib/python3.7/site-packages/celery/result.py", line 342, in maybe_throw
    self.throw(value, self._to_remote_traceback(tb))
  File "/usr/local/lib/python3.7/site-packages/celery/result.py", line 335, in throw
    self.on_ready.throw(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/vine/promises.py", line 244, in throw
    reraise(type(exc), exc, tb)
  File "/usr/local/lib/python3.7/site-packages/vine/five.py", line 195, in reraise
    raise value
Exception: <class 'sqlalchemy.exc.IntegrityError'>(('(psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "pg_type_typname_nsp_index"\nDETAIL:  Key (typname, typnamespace)=(taskset_id_sequence, 2200) already exists.\n',))
```

One workaround is  to force the table creation ahead of time as was proposed by a user in the issue I linked: https://github.com/celery/celery/issues/4653#issuecomment-400029147.

I think Celery should handle this itself. A possible solution would catch `IntegrityError` and try again until `create_all` succeeds. (Perhaps with a limited number of retries and with sleeps compared to this snippet):

```python
    def prepare_models(self, engine):
        from sqlalchemy.exc import IntegrityError
        if not self.prepared:
            while True:
                try:
                    ResultModelBase.metadata.create_all(engine)
                except IntegrityError:
                    continue
                else:
                    break
            self.prepared = True
```

## Minimally Reproducible Test Case
<!--
Please provide a reproducible test case.
Refer to the Reporting Bugs section in our contribution guide.

We prefer submitting test cases in the form of a PR to our integration test suite.
If you can provide one, please mention the PR number below.
If not, please attach the most minimal code example required to reproduce the issue below.
If the test case is too large, please include a link to a gist or a repository below.
-->

This example doesn't use celery at all, but shows that calling create_all in multiple processes can cause the error. It's a race condition, so you might need to try it multiple times or play around with the number of processes:

<details>
<p>


Requires a local postgres, and this database must be created:

```
createdb racetest
```

```python
from concurrent.futures import ProcessPoolExecutor, as_completed

from sqlalchemy import Column, Integer, Table, MetaData, create_engine

metadata = MetaData()

tbl1 = Table('tbl1', metadata, Column('id', Integer, primary_key=True))


def create_all(url):
    engine = create_engine(url)
    metadata.create_all(bind=engine)


def main():
    url = 'postgresql:///racetest'
    engine = create_engine(url)
    
    # Make sure schema is empty before we start
    metadata.drop_all(bind=engine)

    with ProcessPoolExecutor(max_workers=50) as executor:
        futures = []
        for _ in range(50):
            future = executor.submit(create_all, url)
            futures.append(future)

        for fut in as_completed(futures):
            fut.result()


if __name__ == '__main__':
    main()

```

</p>
</details>


--- 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/backends/database/session.py`
Content:
```
1 """SQLAlchemy session."""
2 from kombu.utils.compat import register_after_fork
3 from sqlalchemy import create_engine
4 from sqlalchemy.ext.declarative import declarative_base
5 from sqlalchemy.orm import sessionmaker
6 from sqlalchemy.pool import NullPool
7 
8 ResultModelBase = declarative_base()
9 
10 __all__ = ('SessionManager',)
11 
12 
13 def _after_fork_cleanup_session(session):
14     session._after_fork()
15 
16 
17 class SessionManager:
18     """Manage SQLAlchemy sessions."""
19 
20     def __init__(self):
21         self._engines = {}
22         self._sessions = {}
23         self.forked = False
24         self.prepared = False
25         if register_after_fork is not None:
26             register_after_fork(self, _after_fork_cleanup_session)
27 
28     def _after_fork(self):
29         self.forked = True
30 
31     def get_engine(self, dburi, **kwargs):
32         if self.forked:
33             try:
34                 return self._engines[dburi]
35             except KeyError:
36                 engine = self._engines[dburi] = create_engine(dburi, **kwargs)
37                 return engine
38         else:
39             kwargs = {k: v for k, v in kwargs.items() if
40                       not k.startswith('pool')}
41             return create_engine(dburi, poolclass=NullPool, **kwargs)
42 
43     def create_session(self, dburi, short_lived_sessions=False, **kwargs):
44         engine = self.get_engine(dburi, **kwargs)
45         if self.forked:
46             if short_lived_sessions or dburi not in self._sessions:
47                 self._sessions[dburi] = sessionmaker(bind=engine)
48             return engine, self._sessions[dburi]
49         return engine, sessionmaker(bind=engine)
50 
51     def prepare_models(self, engine):
52         if not self.prepared:
53             ResultModelBase.metadata.create_all(engine)
54             self.prepared = True
55 
56     def session_factory(self, dburi, **kwargs):
57         engine, session = self.create_session(dburi, **kwargs)
58         self.prepare_models(engine)
59         return session()
60
```
--- 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/backends/database/session.py b/celery/backends/database/session.py
--- a/celery/backends/database/session.py
+++ b/celery/backends/database/session.py
@@ -1,14 +1,21 @@
 """SQLAlchemy session."""
+import time
+
 from kombu.utils.compat import register_after_fork
 from sqlalchemy import create_engine
+from sqlalchemy.exc import DatabaseError
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.pool import NullPool
 
+from celery.utils.time import get_exponential_backoff_interval
+
 ResultModelBase = declarative_base()
 
 __all__ = ('SessionManager',)
 
+PREPARE_MODELS_MAX_RETRIES = 10
+
 
 def _after_fork_cleanup_session(session):
     session._after_fork()
@@ -50,7 +57,25 @@
 
     def prepare_models(self, engine):
         if not self.prepared:
-            ResultModelBase.metadata.create_all(engine)
+            # SQLAlchemy will check if the items exist before trying to
+            # create them, which is a race condition. If it raises an error
+            # in one iteration, the next may pass all the existence checks
+            # and the call will succeed.
+            retries = 0
+            while True:
+                try:
+                    ResultModelBase.metadata.create_all(engine)
+                except DatabaseError:
+                    if retries < PREPARE_MODELS_MAX_RETRIES:
+                        sleep_amount_ms = get_exponential_backoff_interval(
+                            10, retries, 1000, True
+                        )
+                        time.sleep(sleep_amount_ms / 1000)
+                        retries += 1
+                    else:
+                        raise
+                else:
+                    break
             self.prepared = True
 
     def session_factory(self, dburi, **kwargs):

{"golden_diff": "diff --git a/celery/backends/database/session.py b/celery/backends/database/session.py\n--- a/celery/backends/database/session.py\n+++ b/celery/backends/database/session.py\n@@ -1,14 +1,21 @@\n \"\"\"SQLAlchemy session.\"\"\"\n+import time\n+\n from kombu.utils.compat import register_after_fork\n from sqlalchemy import create_engine\n+from sqlalchemy.exc import DatabaseError\n from sqlalchemy.ext.declarative import declarative_base\n from sqlalchemy.orm import sessionmaker\n from sqlalchemy.pool import NullPool\n \n+from celery.utils.time import get_exponential_backoff_interval\n+\n ResultModelBase = declarative_base()\n \n __all__ = ('SessionManager',)\n \n+PREPARE_MODELS_MAX_RETRIES = 10\n+\n \n def _after_fork_cleanup_session(session):\n     session._after_fork()\n@@ -50,7 +57,25 @@\n \n     def prepare_models(self, engine):\n         if not self.prepared:\n-            ResultModelBase.metadata.create_all(engine)\n+            # SQLAlchemy will check if the items exist before trying to\n+            # create them, which is a race condition. If it raises an error\n+            # in one iteration, the next may pass all the existence checks\n+            # and the call will succeed.\n+            retries = 0\n+            while True:\n+                try:\n+                    ResultModelBase.metadata.create_all(engine)\n+                except DatabaseError:\n+                    if retries < PREPARE_MODELS_MAX_RETRIES:\n+                        sleep_amount_ms = get_exponential_backoff_interval(\n+                            10, retries, 1000, True\n+                        )\n+                        time.sleep(sleep_amount_ms / 1000)\n+                        retries += 1\n+                    else:\n+                        raise\n+                else:\n+                    break\n             self.prepared = True\n \n     def session_factory(self, dburi, **kwargs):\n", "issue": "Database backend race condition during table creation\n<!--\r\nPlease fill this template entirely and do not erase parts of it.\r\nWe reserve the right to close without a response\r\nbug reports which are incomplete.\r\n-->\r\n# Checklist\r\n<!--\r\nTo check an item on the list replace [ ] with [x].\r\n-->\r\n- [x] I have verified that the issue exists against the `master` branch of Celery.\r\n- [ ] This has already been asked to the [discussion group](https://groups.google.com/forum/#!forum/celery-users) first.\r\n- [x] I have read the relevant section in the\r\n  [contribution guide](http://docs.celeryproject.org/en/latest/contributing.html#other-bugs)\r\n  on reporting bugs.\r\n- [x] I have checked the [issues list](https://github.com/celery/celery/issues?q=is%3Aissue+label%3A%22Issue+Type%3A+Bug+Report%22+-label%3A%22Category%3A+Documentation%22)\r\n  for similar or identical bug reports.\r\n- [x] I have checked the [pull requests list](https://github.com/celery/celery/pulls?q=is%3Apr+label%3A%22PR+Type%3A+Bugfix%22+-label%3A%22Category%3A+Documentation%22)\r\n  for existing proposed fixes.\r\n- [x] I have checked the [commit log](https://github.com/celery/celery/commits/master)\r\n  to find out if the bug was already fixed in the master branch.\r\n- [x] I have included all related issues and possible duplicate issues\r\n  in this issue (If there are none, check this box anyway).\r\n\r\n## Mandatory Debugging Information\r\n\r\n- [x] 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\n- [x] I have verified that the issue exists against the `master` branch of Celery.\r\n- [ ] I have included the contents of ``pip freeze`` in the issue. **N/A**\r\n- [ ] I have included all the versions of all the external dependencies required\r\n  to reproduce this bug. **N/A**\r\n\r\n## Optional Debugging Information\r\n<!--\r\nTry some of the below if you think they are relevant.\r\nIt will help us figure out the scope of the bug and how many users it affects.\r\n-->\r\n- [ ] I have tried reproducing the issue on more than one Python version\r\n  and/or implementation.\r\n- [ ] I have tried reproducing the issue on more than one message broker and/or\r\n  result backend.\r\n- [ ] I have tried reproducing the issue on more than one version of the message\r\n  broker and/or result backend.\r\n- [ ] I have tried reproducing the issue on more than one operating system.\r\n- [ ] I have tried reproducing the issue on more than one workers pool.\r\n- [ ] I have tried reproducing the issue with autoscaling, retries,\r\n  ETA/Countdown & rate limits disabled.\r\n- [ ] I have tried reproducing the issue after downgrading\r\n  and/or upgrading Celery and its dependencies.\r\n\r\n## Related Issues and Possible Duplicates\r\n<!--\r\nPlease make sure to search and mention any related issues\r\nor possible duplicates to this issue as requested by the checklist above.\r\n\r\nThis may or may not include issues in other repositories that the Celery project\r\nmaintains or other repositories that are dependencies of Celery.\r\n\r\nIf you don't know how to mention issues, please refer to Github's documentation\r\non the subject: https://help.github.com/en/articles/autolinked-references-and-urls#issues-and-pull-requests\r\n-->\r\n\r\n#### Related Issues\r\n\r\n- https://github.com/celery/celery/issues/4653\r\n\r\n#### Possible Duplicates\r\n\r\n- None\r\n\r\n## Environment & Settings\r\n<!-- Include the contents of celery --version below -->\r\n**Celery version**: 4.4.7 (cliffs)\r\n<!-- Include the output of celery -A proj report below -->\r\n<details>\r\n<summary><b><code>celery report</code> Output:</b></summary>\r\n<p>\r\n\r\n```\r\nsoftware -> celery:4.4.7 (cliffs) kombu:4.6.11 py:3.8.2\r\n            billiard:3.6.3.0 py-amqp:2.6.1\r\nplatform -> system:Darwin arch:64bit\r\n            kernel version:16.7.0 imp:CPython\r\nloader   -> celery.loaders.app.AppLoader\r\nsettings -> transport:amqp results:db+postgresql://<redacted>\r\n\r\ninclude: ['redacted']\r\naccept_content: ['redacted-custom']\r\ndatabase_table_names: {\r\n 'group': 'celery_group', 'task': 'celery_task'}\r\nresult_serializer: 'redacted-custom'\r\ntask_serializer: 'redacted-custom'\r\ntask_track_started: True\r\nbroker_url: 'amqp://<redacted>'\r\nresult_backend: 'db+postgresql://<redacted>'\r\n```\r\n\r\n</p>\r\n</details>\r\n\r\n# Steps to Reproduce\r\n\r\nWhen celery uses a database result backend, the following line can be called multiple times from different processes:\r\n\r\nhttps://github.com/celery/celery/blob/9a6c2923e859b6993227605610255bd632c1ae68/celery/backends/database/session.py#L56\r\n\r\nThis is a race condition because SQLAlchemy first checks if the tables/sequences exist and then tries to create them. It causes errors like this (at least on PostgreSQL):\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.7/site-packages/redacted.py\", line 168, in _redacted\r\n    result = async_result.get()\r\n  File \"/usr/local/lib/python3.7/site-packages/celery/result.py\", line 226, in get\r\n    self.maybe_throw(callback=callback)\r\n  File \"/usr/local/lib/python3.7/site-packages/celery/result.py\", line 342, in maybe_throw\r\n    self.throw(value, self._to_remote_traceback(tb))\r\n  File \"/usr/local/lib/python3.7/site-packages/celery/result.py\", line 335, in throw\r\n    self.on_ready.throw(*args, **kwargs)\r\n  File \"/usr/local/lib/python3.7/site-packages/vine/promises.py\", line 244, in throw\r\n    reraise(type(exc), exc, tb)\r\n  File \"/usr/local/lib/python3.7/site-packages/vine/five.py\", line 195, in reraise\r\n    raise value\r\nException: <class 'sqlalchemy.exc.IntegrityError'>(('(psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint \"pg_type_typname_nsp_index\"\\nDETAIL:  Key (typname, typnamespace)=(taskset_id_sequence, 2200) already exists.\\n',))\r\n```\r\n\r\nOne workaround is  to force the table creation ahead of time as was proposed by a user in the issue I linked: https://github.com/celery/celery/issues/4653#issuecomment-400029147.\r\n\r\nI think Celery should handle this itself. A possible solution would catch `IntegrityError` and try again until `create_all` succeeds. (Perhaps with a limited number of retries and with sleeps compared to this snippet):\r\n\r\n```python\r\n    def prepare_models(self, engine):\r\n        from sqlalchemy.exc import IntegrityError\r\n        if not self.prepared:\r\n            while True:\r\n                try:\r\n                    ResultModelBase.metadata.create_all(engine)\r\n                except IntegrityError:\r\n                    continue\r\n                else:\r\n                    break\r\n            self.prepared = True\r\n```\r\n\r\n## Minimally Reproducible Test Case\r\n<!--\r\nPlease provide a reproducible test case.\r\nRefer to the Reporting Bugs section in our contribution guide.\r\n\r\nWe prefer submitting test cases in the form of a PR to our integration test suite.\r\nIf you can provide one, please mention the PR number below.\r\nIf not, please attach the most minimal code example required to reproduce the issue below.\r\nIf the test case is too large, please include a link to a gist or a repository below.\r\n-->\r\n\r\nThis example doesn't use celery at all, but shows that calling create_all in multiple processes can cause the error. It's a race condition, so you might need to try it multiple times or play around with the number of processes:\r\n\r\n<details>\r\n<p>\r\n\r\n\r\nRequires a local postgres, and this database must be created:\r\n\r\n```\r\ncreatedb racetest\r\n```\r\n\r\n```python\r\nfrom concurrent.futures import ProcessPoolExecutor, as_completed\r\n\r\nfrom sqlalchemy import Column, Integer, Table, MetaData, create_engine\r\n\r\nmetadata = MetaData()\r\n\r\ntbl1 = Table('tbl1', metadata, Column('id', Integer, primary_key=True))\r\n\r\n\r\ndef create_all(url):\r\n    engine = create_engine(url)\r\n    metadata.create_all(bind=engine)\r\n\r\n\r\ndef main():\r\n    url = 'postgresql:///racetest'\r\n    engine = create_engine(url)\r\n    \r\n    # Make sure schema is empty before we start\r\n    metadata.drop_all(bind=engine)\r\n\r\n    with ProcessPoolExecutor(max_workers=50) as executor:\r\n        futures = []\r\n        for _ in range(50):\r\n            future = executor.submit(create_all, url)\r\n            futures.append(future)\r\n\r\n        for fut in as_completed(futures):\r\n            fut.result()\r\n\r\n\r\nif __name__ == '__main__':\r\n    main()\r\n\r\n```\r\n\r\n</p>\r\n</details>\r\n\n", "before_files": [{"content": "\"\"\"SQLAlchemy session.\"\"\"\nfrom kombu.utils.compat import register_after_fork\nfrom sqlalchemy import create_engine\nfrom sqlalchemy.ext.declarative import declarative_base\nfrom sqlalchemy.orm import sessionmaker\nfrom sqlalchemy.pool import NullPool\n\nResultModelBase = declarative_base()\n\n__all__ = ('SessionManager',)\n\n\ndef _after_fork_cleanup_session(session):\n    session._after_fork()\n\n\nclass SessionManager:\n    \"\"\"Manage SQLAlchemy sessions.\"\"\"\n\n    def __init__(self):\n        self._engines = {}\n        self._sessions = {}\n        self.forked = False\n        self.prepared = False\n        if register_after_fork is not None:\n            register_after_fork(self, _after_fork_cleanup_session)\n\n    def _after_fork(self):\n        self.forked = True\n\n    def get_engine(self, dburi, **kwargs):\n        if self.forked:\n            try:\n                return self._engines[dburi]\n            except KeyError:\n                engine = self._engines[dburi] = create_engine(dburi, **kwargs)\n                return engine\n        else:\n            kwargs = {k: v for k, v in kwargs.items() if\n                      not k.startswith('pool')}\n            return create_engine(dburi, poolclass=NullPool, **kwargs)\n\n    def create_session(self, dburi, short_lived_sessions=False, **kwargs):\n        engine = self.get_engine(dburi, **kwargs)\n        if self.forked:\n            if short_lived_sessions or dburi not in self._sessions:\n                self._sessions[dburi] = sessionmaker(bind=engine)\n            return engine, self._sessions[dburi]\n        return engine, sessionmaker(bind=engine)\n\n    def prepare_models(self, engine):\n        if not self.prepared:\n            ResultModelBase.metadata.create_all(engine)\n            self.prepared = True\n\n    def session_factory(self, dburi, **kwargs):\n        engine, session = self.create_session(dburi, **kwargs)\n        self.prepare_models(engine)\n        return session()\n", "path": "celery/backends/database/session.py"}], "after_files": [{"content": "\"\"\"SQLAlchemy session.\"\"\"\nimport time\n\nfrom kombu.utils.compat import register_after_fork\nfrom sqlalchemy import create_engine\nfrom sqlalchemy.exc import DatabaseError\nfrom sqlalchemy.ext.declarative import declarative_base\nfrom sqlalchemy.orm import sessionmaker\nfrom sqlalchemy.pool import NullPool\n\nfrom celery.utils.time import get_exponential_backoff_interval\n\nResultModelBase = declarative_base()\n\n__all__ = ('SessionManager',)\n\nPREPARE_MODELS_MAX_RETRIES = 10\n\n\ndef _after_fork_cleanup_session(session):\n    session._after_fork()\n\n\nclass SessionManager:\n    \"\"\"Manage SQLAlchemy sessions.\"\"\"\n\n    def __init__(self):\n        self._engines = {}\n        self._sessions = {}\n        self.forked = False\n        self.prepared = False\n        if register_after_fork is not None:\n            register_after_fork(self, _after_fork_cleanup_session)\n\n    def _after_fork(self):\n        self.forked = True\n\n    def get_engine(self, dburi, **kwargs):\n        if self.forked:\n            try:\n                return self._engines[dburi]\n            except KeyError:\n                engine = self._engines[dburi] = create_engine(dburi, **kwargs)\n                return engine\n        else:\n            kwargs = {k: v for k, v in kwargs.items() if\n                      not k.startswith('pool')}\n            return create_engine(dburi, poolclass=NullPool, **kwargs)\n\n    def create_session(self, dburi, short_lived_sessions=False, **kwargs):\n        engine = self.get_engine(dburi, **kwargs)\n        if self.forked:\n            if short_lived_sessions or dburi not in self._sessions:\n                self._sessions[dburi] = sessionmaker(bind=engine)\n            return engine, self._sessions[dburi]\n        return engine, sessionmaker(bind=engine)\n\n    def prepare_models(self, engine):\n        if not self.prepared:\n            # SQLAlchemy will check if the items exist before trying to\n            # create them, which is a race condition. If it raises an error\n            # in one iteration, the next may pass all the existence checks\n            # and the call will succeed.\n            retries = 0\n            while True:\n                try:\n                    ResultModelBase.metadata.create_all(engine)\n                except DatabaseError:\n                    if retries < PREPARE_MODELS_MAX_RETRIES:\n                        sleep_amount_ms = get_exponential_backoff_interval(\n                            10, retries, 1000, True\n                        )\n                        time.sleep(sleep_amount_ms / 1000)\n                        retries += 1\n                    else:\n                        raise\n                else:\n                    break\n            self.prepared = True\n\n    def session_factory(self, dburi, **kwargs):\n        engine, session = self.create_session(dburi, **kwargs)\n        self.prepare_models(engine)\n        return session()\n", "path": "celery/backends/database/session.py"}]}