We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
DatabaseBackend._update_result()  have an error property.
python 3.7
celery 4.4.0rc3 

The result has an error value NULL for the name in my backend(mysql), but it's work well when I use redis as my backend.

After I change this error in `backends/database/__init__.py` [135], alter 'task_name' to 'task', I get the correct task_name.

The 'name' in `backends/base.py` [706,717]
```
        if self.app.conf.find_value_for_key('extended', 'result'):
            if request:
                request_meta = {
->                  'name': getattr(request, 'task', None),
                    'args': getattr(request, 'args', None),
                    'kwargs': getattr(request, 'kwargs', None),
                    'worker': getattr(request, 'hostname', None),
                    'retries': getattr(request, 'retries', None),
                    'queue': request.delivery_info.get('routing_key')
                    if hasattr(request, 'delivery_info') and
                    request.delivery_info else None
                }
```
The 'name' in `backends/database/__init__.py` [129,148]
```
    def _update_result(self, task, result, state, traceback=None,
                       request=None):
        task.result = result
        task.status = state
        task.traceback = traceback
        if self.app.conf.find_value_for_key('extended', 'result'):
-           task.name = getattr(request, 'task_name', None)
+           task.name = getattr(request, 'task', None)
            task.args = ensure_bytes(
                self.encode(getattr(request, 'args', None))
            )
            task.kwargs = ensure_bytes(
                self.encode(getattr(request, 'kwargs', None))
            )
            task.worker = getattr(request, 'hostname', None)
            task.retries = getattr(request, 'retries', None)
            task.queue = (
                request.delivery_info.get("routing_key")
                if hasattr(request, "delivery_info") and request.delivery_info
                else None
            )
```


--- 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/__init__.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 """SQLAlchemy result store backend."""
3 from __future__ import absolute_import, unicode_literals
4 
5 import logging
6 from contextlib import contextmanager
7 
8 from kombu.utils.encoding import ensure_bytes
9 from vine.utils import wraps
10 
11 from celery import states
12 from celery.backends.base import BaseBackend
13 from celery.exceptions import ImproperlyConfigured
14 from celery.five import range
15 from celery.utils.time import maybe_timedelta
16 
17 from .models import Task, TaskExtended, TaskSet
18 from .session import SessionManager
19 
20 try:
21     from sqlalchemy.exc import DatabaseError, InvalidRequestError
22     from sqlalchemy.orm.exc import StaleDataError
23 except ImportError:  # pragma: no cover
24     raise ImproperlyConfigured(
25         'The database result backend requires SQLAlchemy to be installed.'
26         'See https://pypi.org/project/SQLAlchemy/')
27 
28 logger = logging.getLogger(__name__)
29 
30 __all__ = ('DatabaseBackend',)
31 
32 
33 @contextmanager
34 def session_cleanup(session):
35     try:
36         yield
37     except Exception:
38         session.rollback()
39         raise
40     finally:
41         session.close()
42 
43 
44 def retry(fun):
45 
46     @wraps(fun)
47     def _inner(*args, **kwargs):
48         max_retries = kwargs.pop('max_retries', 3)
49 
50         for retries in range(max_retries):
51             try:
52                 return fun(*args, **kwargs)
53             except (DatabaseError, InvalidRequestError, StaleDataError):
54                 logger.warning(
55                     'Failed operation %s.  Retrying %s more times.',
56                     fun.__name__, max_retries - retries - 1,
57                     exc_info=True)
58                 if retries + 1 >= max_retries:
59                     raise
60 
61     return _inner
62 
63 
64 class DatabaseBackend(BaseBackend):
65     """The database result backend."""
66 
67     # ResultSet.iterate should sleep this much between each pool,
68     # to not bombard the database with queries.
69     subpolling_interval = 0.5
70 
71     task_cls = Task
72     taskset_cls = TaskSet
73 
74     def __init__(self, dburi=None, engine_options=None, url=None, **kwargs):
75         # The `url` argument was added later and is used by
76         # the app to set backend by url (celery.app.backends.by_url)
77         super(DatabaseBackend, self).__init__(expires_type=maybe_timedelta,
78                                               url=url, **kwargs)
79         conf = self.app.conf
80 
81         if self.extended_result:
82             self.task_cls = TaskExtended
83 
84         self.url = url or dburi or conf.database_url
85         self.engine_options = dict(
86             engine_options or {},
87             **conf.database_engine_options or {})
88         self.short_lived_sessions = kwargs.get(
89             'short_lived_sessions',
90             conf.database_short_lived_sessions)
91 
92         tablenames = conf.database_table_names or {}
93         self.task_cls.__table__.name = tablenames.get('task',
94                                                       'celery_taskmeta')
95         self.taskset_cls.__table__.name = tablenames.get('group',
96                                                          'celery_tasksetmeta')
97 
98         if not self.url:
99             raise ImproperlyConfigured(
100                 'Missing connection string! Do you have the'
101                 ' database_url setting set to a real value?')
102 
103     @property
104     def extended_result(self):
105         return self.app.conf.find_value_for_key('extended', 'result')
106 
107     def ResultSession(self, session_manager=SessionManager()):
108         return session_manager.session_factory(
109             dburi=self.url,
110             short_lived_sessions=self.short_lived_sessions,
111             **self.engine_options)
112 
113     @retry
114     def _store_result(self, task_id, result, state, traceback=None,
115                       request=None, **kwargs):
116         """Store return value and state of an executed task."""
117         session = self.ResultSession()
118         with session_cleanup(session):
119             task = list(session.query(self.task_cls).filter(self.task_cls.task_id == task_id))
120             task = task and task[0]
121             if not task:
122                 task = self.task_cls(task_id)
123                 session.add(task)
124                 session.flush()
125 
126             self._update_result(task, result, state, traceback=traceback, request=request)
127             session.commit()
128 
129     def _update_result(self, task, result, state, traceback=None,
130                        request=None):
131         task.result = result
132         task.status = state
133         task.traceback = traceback
134         if self.app.conf.find_value_for_key('extended', 'result'):
135             task.name = getattr(request, 'task_name', None)
136             task.args = ensure_bytes(
137                 self.encode(getattr(request, 'args', None))
138             )
139             task.kwargs = ensure_bytes(
140                 self.encode(getattr(request, 'kwargs', None))
141             )
142             task.worker = getattr(request, 'hostname', None)
143             task.retries = getattr(request, 'retries', None)
144             task.queue = (
145                 request.delivery_info.get("routing_key")
146                 if hasattr(request, "delivery_info") and request.delivery_info
147                 else None
148             )
149 
150     @retry
151     def _get_task_meta_for(self, task_id):
152         """Get task meta-data for a task by id."""
153         session = self.ResultSession()
154         with session_cleanup(session):
155             task = list(session.query(self.task_cls).filter(self.task_cls.task_id == task_id))
156             task = task and task[0]
157             if not task:
158                 task = self.task_cls(task_id)
159                 task.status = states.PENDING
160                 task.result = None
161             data = task.to_dict()
162             if 'args' in data:
163                 data['args'] = self.decode(data['args'])
164             if 'kwargs' in data:
165                 data['kwargs'] = self.decode(data['kwargs'])
166             return self.meta_from_decoded(data)
167 
168     @retry
169     def _save_group(self, group_id, result):
170         """Store the result of an executed group."""
171         session = self.ResultSession()
172         with session_cleanup(session):
173             group = self.taskset_cls(group_id, result)
174             session.add(group)
175             session.flush()
176             session.commit()
177             return result
178 
179     @retry
180     def _restore_group(self, group_id):
181         """Get meta-data for group by id."""
182         session = self.ResultSession()
183         with session_cleanup(session):
184             group = session.query(self.taskset_cls).filter(
185                 self.taskset_cls.taskset_id == group_id).first()
186             if group:
187                 return group.to_dict()
188 
189     @retry
190     def _delete_group(self, group_id):
191         """Delete meta-data for group by id."""
192         session = self.ResultSession()
193         with session_cleanup(session):
194             session.query(self.taskset_cls).filter(
195                 self.taskset_cls.taskset_id == group_id).delete()
196             session.flush()
197             session.commit()
198 
199     @retry
200     def _forget(self, task_id):
201         """Forget about result."""
202         session = self.ResultSession()
203         with session_cleanup(session):
204             session.query(self.task_cls).filter(self.task_cls.task_id == task_id).delete()
205             session.commit()
206 
207     def cleanup(self):
208         """Delete expired meta-data."""
209         session = self.ResultSession()
210         expires = self.expires
211         now = self.app.now()
212         with session_cleanup(session):
213             session.query(self.task_cls).filter(
214                 self.task_cls.date_done < (now - expires)).delete()
215             session.query(self.taskset_cls).filter(
216                 self.taskset_cls.date_done < (now - expires)).delete()
217             session.commit()
218 
219     def __reduce__(self, args=(), kwargs=None):
220         kwargs = {} if not kwargs else kwargs
221         kwargs.update(
222             {'dburi': self.url,
223              'expires': self.expires,
224              'engine_options': self.engine_options})
225         return super(DatabaseBackend, self).__reduce__(args, kwargs)
226
```
--- 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/__init__.py b/celery/backends/database/__init__.py
--- a/celery/backends/database/__init__.py
+++ b/celery/backends/database/__init__.py
@@ -132,7 +132,7 @@
         task.status = state
         task.traceback = traceback
         if self.app.conf.find_value_for_key('extended', 'result'):
-            task.name = getattr(request, 'task_name', None)
+            task.name = getattr(request, 'task', None)
             task.args = ensure_bytes(
                 self.encode(getattr(request, 'args', None))
             )

{"golden_diff": "diff --git a/celery/backends/database/__init__.py b/celery/backends/database/__init__.py\n--- a/celery/backends/database/__init__.py\n+++ b/celery/backends/database/__init__.py\n@@ -132,7 +132,7 @@\n         task.status = state\n         task.traceback = traceback\n         if self.app.conf.find_value_for_key('extended', 'result'):\n-            task.name = getattr(request, 'task_name', None)\n+            task.name = getattr(request, 'task', None)\n             task.args = ensure_bytes(\n                 self.encode(getattr(request, 'args', None))\n             )\n", "issue": "DatabaseBackend._update_result()  have an error property.\npython 3.7\r\ncelery 4.4.0rc3 \r\n\r\nThe result has an error value NULL for the name in my backend(mysql), but it's work well when I use redis as my backend.\r\n\r\nAfter I change this error in `backends/database/__init__.py` [135], alter 'task_name' to 'task', I get the correct task_name.\r\n\r\nThe 'name' in `backends/base.py` [706,717]\r\n```\r\n        if self.app.conf.find_value_for_key('extended', 'result'):\r\n            if request:\r\n                request_meta = {\r\n->                  'name': getattr(request, 'task', None),\r\n                    'args': getattr(request, 'args', None),\r\n                    'kwargs': getattr(request, 'kwargs', None),\r\n                    'worker': getattr(request, 'hostname', None),\r\n                    'retries': getattr(request, 'retries', None),\r\n                    'queue': request.delivery_info.get('routing_key')\r\n                    if hasattr(request, 'delivery_info') and\r\n                    request.delivery_info else None\r\n                }\r\n```\r\nThe 'name' in `backends/database/__init__.py` [129,148]\r\n```\r\n    def _update_result(self, task, result, state, traceback=None,\r\n                       request=None):\r\n        task.result = result\r\n        task.status = state\r\n        task.traceback = traceback\r\n        if self.app.conf.find_value_for_key('extended', 'result'):\r\n-           task.name = getattr(request, 'task_name', None)\r\n+           task.name = getattr(request, 'task', None)\r\n            task.args = ensure_bytes(\r\n                self.encode(getattr(request, 'args', None))\r\n            )\r\n            task.kwargs = ensure_bytes(\r\n                self.encode(getattr(request, 'kwargs', None))\r\n            )\r\n            task.worker = getattr(request, 'hostname', None)\r\n            task.retries = getattr(request, 'retries', None)\r\n            task.queue = (\r\n                request.delivery_info.get(\"routing_key\")\r\n                if hasattr(request, \"delivery_info\") and request.delivery_info\r\n                else None\r\n            )\r\n```\r\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"SQLAlchemy result store backend.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport logging\nfrom contextlib import contextmanager\n\nfrom kombu.utils.encoding import ensure_bytes\nfrom vine.utils import wraps\n\nfrom celery import states\nfrom celery.backends.base import BaseBackend\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import range\nfrom celery.utils.time import maybe_timedelta\n\nfrom .models import Task, TaskExtended, TaskSet\nfrom .session import SessionManager\n\ntry:\n    from sqlalchemy.exc import DatabaseError, InvalidRequestError\n    from sqlalchemy.orm.exc import StaleDataError\nexcept ImportError:  # pragma: no cover\n    raise ImproperlyConfigured(\n        'The database result backend requires SQLAlchemy to be installed.'\n        'See https://pypi.org/project/SQLAlchemy/')\n\nlogger = logging.getLogger(__name__)\n\n__all__ = ('DatabaseBackend',)\n\n\n@contextmanager\ndef session_cleanup(session):\n    try:\n        yield\n    except Exception:\n        session.rollback()\n        raise\n    finally:\n        session.close()\n\n\ndef retry(fun):\n\n    @wraps(fun)\n    def _inner(*args, **kwargs):\n        max_retries = kwargs.pop('max_retries', 3)\n\n        for retries in range(max_retries):\n            try:\n                return fun(*args, **kwargs)\n            except (DatabaseError, InvalidRequestError, StaleDataError):\n                logger.warning(\n                    'Failed operation %s.  Retrying %s more times.',\n                    fun.__name__, max_retries - retries - 1,\n                    exc_info=True)\n                if retries + 1 >= max_retries:\n                    raise\n\n    return _inner\n\n\nclass DatabaseBackend(BaseBackend):\n    \"\"\"The database result backend.\"\"\"\n\n    # ResultSet.iterate should sleep this much between each pool,\n    # to not bombard the database with queries.\n    subpolling_interval = 0.5\n\n    task_cls = Task\n    taskset_cls = TaskSet\n\n    def __init__(self, dburi=None, engine_options=None, url=None, **kwargs):\n        # The `url` argument was added later and is used by\n        # the app to set backend by url (celery.app.backends.by_url)\n        super(DatabaseBackend, self).__init__(expires_type=maybe_timedelta,\n                                              url=url, **kwargs)\n        conf = self.app.conf\n\n        if self.extended_result:\n            self.task_cls = TaskExtended\n\n        self.url = url or dburi or conf.database_url\n        self.engine_options = dict(\n            engine_options or {},\n            **conf.database_engine_options or {})\n        self.short_lived_sessions = kwargs.get(\n            'short_lived_sessions',\n            conf.database_short_lived_sessions)\n\n        tablenames = conf.database_table_names or {}\n        self.task_cls.__table__.name = tablenames.get('task',\n                                                      'celery_taskmeta')\n        self.taskset_cls.__table__.name = tablenames.get('group',\n                                                         'celery_tasksetmeta')\n\n        if not self.url:\n            raise ImproperlyConfigured(\n                'Missing connection string! Do you have the'\n                ' database_url setting set to a real value?')\n\n    @property\n    def extended_result(self):\n        return self.app.conf.find_value_for_key('extended', 'result')\n\n    def ResultSession(self, session_manager=SessionManager()):\n        return session_manager.session_factory(\n            dburi=self.url,\n            short_lived_sessions=self.short_lived_sessions,\n            **self.engine_options)\n\n    @retry\n    def _store_result(self, task_id, result, state, traceback=None,\n                      request=None, **kwargs):\n        \"\"\"Store return value and state of an executed task.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            task = list(session.query(self.task_cls).filter(self.task_cls.task_id == task_id))\n            task = task and task[0]\n            if not task:\n                task = self.task_cls(task_id)\n                session.add(task)\n                session.flush()\n\n            self._update_result(task, result, state, traceback=traceback, request=request)\n            session.commit()\n\n    def _update_result(self, task, result, state, traceback=None,\n                       request=None):\n        task.result = result\n        task.status = state\n        task.traceback = traceback\n        if self.app.conf.find_value_for_key('extended', 'result'):\n            task.name = getattr(request, 'task_name', None)\n            task.args = ensure_bytes(\n                self.encode(getattr(request, 'args', None))\n            )\n            task.kwargs = ensure_bytes(\n                self.encode(getattr(request, 'kwargs', None))\n            )\n            task.worker = getattr(request, 'hostname', None)\n            task.retries = getattr(request, 'retries', None)\n            task.queue = (\n                request.delivery_info.get(\"routing_key\")\n                if hasattr(request, \"delivery_info\") and request.delivery_info\n                else None\n            )\n\n    @retry\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task meta-data for a task by id.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            task = list(session.query(self.task_cls).filter(self.task_cls.task_id == task_id))\n            task = task and task[0]\n            if not task:\n                task = self.task_cls(task_id)\n                task.status = states.PENDING\n                task.result = None\n            data = task.to_dict()\n            if 'args' in data:\n                data['args'] = self.decode(data['args'])\n            if 'kwargs' in data:\n                data['kwargs'] = self.decode(data['kwargs'])\n            return self.meta_from_decoded(data)\n\n    @retry\n    def _save_group(self, group_id, result):\n        \"\"\"Store the result of an executed group.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            group = self.taskset_cls(group_id, result)\n            session.add(group)\n            session.flush()\n            session.commit()\n            return result\n\n    @retry\n    def _restore_group(self, group_id):\n        \"\"\"Get meta-data for group by id.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            group = session.query(self.taskset_cls).filter(\n                self.taskset_cls.taskset_id == group_id).first()\n            if group:\n                return group.to_dict()\n\n    @retry\n    def _delete_group(self, group_id):\n        \"\"\"Delete meta-data for group by id.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            session.query(self.taskset_cls).filter(\n                self.taskset_cls.taskset_id == group_id).delete()\n            session.flush()\n            session.commit()\n\n    @retry\n    def _forget(self, task_id):\n        \"\"\"Forget about result.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            session.query(self.task_cls).filter(self.task_cls.task_id == task_id).delete()\n            session.commit()\n\n    def cleanup(self):\n        \"\"\"Delete expired meta-data.\"\"\"\n        session = self.ResultSession()\n        expires = self.expires\n        now = self.app.now()\n        with session_cleanup(session):\n            session.query(self.task_cls).filter(\n                self.task_cls.date_done < (now - expires)).delete()\n            session.query(self.taskset_cls).filter(\n                self.taskset_cls.date_done < (now - expires)).delete()\n            session.commit()\n\n    def __reduce__(self, args=(), kwargs=None):\n        kwargs = {} if not kwargs else kwargs\n        kwargs.update(\n            {'dburi': self.url,\n             'expires': self.expires,\n             'engine_options': self.engine_options})\n        return super(DatabaseBackend, self).__reduce__(args, kwargs)\n", "path": "celery/backends/database/__init__.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"SQLAlchemy result store backend.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport logging\nfrom contextlib import contextmanager\n\nfrom kombu.utils.encoding import ensure_bytes\nfrom vine.utils import wraps\n\nfrom celery import states\nfrom celery.backends.base import BaseBackend\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import range\nfrom celery.utils.time import maybe_timedelta\n\nfrom .models import Task, TaskExtended, TaskSet\nfrom .session import SessionManager\n\ntry:\n    from sqlalchemy.exc import DatabaseError, InvalidRequestError\n    from sqlalchemy.orm.exc import StaleDataError\nexcept ImportError:  # pragma: no cover\n    raise ImproperlyConfigured(\n        'The database result backend requires SQLAlchemy to be installed.'\n        'See https://pypi.org/project/SQLAlchemy/')\n\nlogger = logging.getLogger(__name__)\n\n__all__ = ('DatabaseBackend',)\n\n\n@contextmanager\ndef session_cleanup(session):\n    try:\n        yield\n    except Exception:\n        session.rollback()\n        raise\n    finally:\n        session.close()\n\n\ndef retry(fun):\n\n    @wraps(fun)\n    def _inner(*args, **kwargs):\n        max_retries = kwargs.pop('max_retries', 3)\n\n        for retries in range(max_retries):\n            try:\n                return fun(*args, **kwargs)\n            except (DatabaseError, InvalidRequestError, StaleDataError):\n                logger.warning(\n                    'Failed operation %s.  Retrying %s more times.',\n                    fun.__name__, max_retries - retries - 1,\n                    exc_info=True)\n                if retries + 1 >= max_retries:\n                    raise\n\n    return _inner\n\n\nclass DatabaseBackend(BaseBackend):\n    \"\"\"The database result backend.\"\"\"\n\n    # ResultSet.iterate should sleep this much between each pool,\n    # to not bombard the database with queries.\n    subpolling_interval = 0.5\n\n    task_cls = Task\n    taskset_cls = TaskSet\n\n    def __init__(self, dburi=None, engine_options=None, url=None, **kwargs):\n        # The `url` argument was added later and is used by\n        # the app to set backend by url (celery.app.backends.by_url)\n        super(DatabaseBackend, self).__init__(expires_type=maybe_timedelta,\n                                              url=url, **kwargs)\n        conf = self.app.conf\n\n        if self.extended_result:\n            self.task_cls = TaskExtended\n\n        self.url = url or dburi or conf.database_url\n        self.engine_options = dict(\n            engine_options or {},\n            **conf.database_engine_options or {})\n        self.short_lived_sessions = kwargs.get(\n            'short_lived_sessions',\n            conf.database_short_lived_sessions)\n\n        tablenames = conf.database_table_names or {}\n        self.task_cls.__table__.name = tablenames.get('task',\n                                                      'celery_taskmeta')\n        self.taskset_cls.__table__.name = tablenames.get('group',\n                                                         'celery_tasksetmeta')\n\n        if not self.url:\n            raise ImproperlyConfigured(\n                'Missing connection string! Do you have the'\n                ' database_url setting set to a real value?')\n\n    @property\n    def extended_result(self):\n        return self.app.conf.find_value_for_key('extended', 'result')\n\n    def ResultSession(self, session_manager=SessionManager()):\n        return session_manager.session_factory(\n            dburi=self.url,\n            short_lived_sessions=self.short_lived_sessions,\n            **self.engine_options)\n\n    @retry\n    def _store_result(self, task_id, result, state, traceback=None,\n                      request=None, **kwargs):\n        \"\"\"Store return value and state of an executed task.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            task = list(session.query(self.task_cls).filter(self.task_cls.task_id == task_id))\n            task = task and task[0]\n            if not task:\n                task = self.task_cls(task_id)\n                session.add(task)\n                session.flush()\n\n            self._update_result(task, result, state, traceback=traceback, request=request)\n            session.commit()\n\n    def _update_result(self, task, result, state, traceback=None,\n                       request=None):\n        task.result = result\n        task.status = state\n        task.traceback = traceback\n        if self.app.conf.find_value_for_key('extended', 'result'):\n            task.name = getattr(request, 'task', None)\n            task.args = ensure_bytes(\n                self.encode(getattr(request, 'args', None))\n            )\n            task.kwargs = ensure_bytes(\n                self.encode(getattr(request, 'kwargs', None))\n            )\n            task.worker = getattr(request, 'hostname', None)\n            task.retries = getattr(request, 'retries', None)\n            task.queue = (\n                request.delivery_info.get(\"routing_key\")\n                if hasattr(request, \"delivery_info\") and request.delivery_info\n                else None\n            )\n\n    @retry\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task meta-data for a task by id.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            task = list(session.query(self.task_cls).filter(self.task_cls.task_id == task_id))\n            task = task and task[0]\n            if not task:\n                task = self.task_cls(task_id)\n                task.status = states.PENDING\n                task.result = None\n            data = task.to_dict()\n            if 'args' in data:\n                data['args'] = self.decode(data['args'])\n            if 'kwargs' in data:\n                data['kwargs'] = self.decode(data['kwargs'])\n            return self.meta_from_decoded(data)\n\n    @retry\n    def _save_group(self, group_id, result):\n        \"\"\"Store the result of an executed group.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            group = self.taskset_cls(group_id, result)\n            session.add(group)\n            session.flush()\n            session.commit()\n            return result\n\n    @retry\n    def _restore_group(self, group_id):\n        \"\"\"Get meta-data for group by id.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            group = session.query(self.taskset_cls).filter(\n                self.taskset_cls.taskset_id == group_id).first()\n            if group:\n                return group.to_dict()\n\n    @retry\n    def _delete_group(self, group_id):\n        \"\"\"Delete meta-data for group by id.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            session.query(self.taskset_cls).filter(\n                self.taskset_cls.taskset_id == group_id).delete()\n            session.flush()\n            session.commit()\n\n    @retry\n    def _forget(self, task_id):\n        \"\"\"Forget about result.\"\"\"\n        session = self.ResultSession()\n        with session_cleanup(session):\n            session.query(self.task_cls).filter(self.task_cls.task_id == task_id).delete()\n            session.commit()\n\n    def cleanup(self):\n        \"\"\"Delete expired meta-data.\"\"\"\n        session = self.ResultSession()\n        expires = self.expires\n        now = self.app.now()\n        with session_cleanup(session):\n            session.query(self.task_cls).filter(\n                self.task_cls.date_done < (now - expires)).delete()\n            session.query(self.taskset_cls).filter(\n                self.taskset_cls.date_done < (now - expires)).delete()\n            session.commit()\n\n    def __reduce__(self, args=(), kwargs=None):\n        kwargs = {} if not kwargs else kwargs\n        kwargs.update(\n            {'dburi': self.url,\n             'expires': self.expires,\n             'engine_options': self.engine_options})\n        return super(DatabaseBackend, self).__reduce__(args, kwargs)\n", "path": "celery/backends/database/__init__.py"}]}