We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Oracle database backend throws exception
When celery is used with oracle(11g) as database backend. There is an exception thrown

```
[2015-04-29 06:48:36,608: DEBUG/MainProcess] (0.011) QUERY = 'SELECT * FROM (SELECT ROWNUM AS "_RN", "_SUB".* FROM (SELECT "DJKOMBU_MESSAGE"."ID", "DJKOMBU_MESSAGE"."VISIBLE", "DJKOMBU_MESSAGE"."SENT_AT", "DJKOMBU_MESSAGE"."PAYLOAD", "DJKOMBU_MESSAGE"."QUEUE_ID" FROM "DJKOMBU_MESSAGE" WHERE ("DJKOMBU_MESSAGE"."QUEUE_ID" = :arg0 AND "DJKOMBU_MESSAGE"."VISIBLE" = :arg1) ORDER BY "DJKOMBU_MESSAGE"."SENT_AT" ASC, "DJKOMBU_MESSAGE"."ID" ASC FOR UPDATE) "_SUB" WHERE ROWNUM <= 1) WHERE "_RN" > 0' - PARAMS = (1, True); args=(1, True)
[2015-04-29 06:48:36,617: ERROR/MainProcess] Unrecoverable error: DatabaseError(<cx_Oracle._Error object at 0x111105878>,)
Traceback (most recent call last):
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/utils.py", line 65, in execute
    return self.cursor.execute(sql, params)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/oracle/base.py", line 916, in execute
    return self.cursor.execute(query, self._param_generator(params))
cx_Oracle.DatabaseError: ORA-00907: missing right parenthesis


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/worker/__init__.py", line 206, in start
    self.blueprint.start(self)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/bootsteps.py", line 123, in start
    step.start(parent)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/bootsteps.py", line 374, in start
    return self.obj.start()
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/worker/consumer.py", line 278, in start
    blueprint.start(self)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/bootsteps.py", line 123, in start
    step.start(parent)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/worker/consumer.py", line 821, in start
    c.loop(*c.loop_args())
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/worker/loops.py", line 97, in synloop
    connection.drain_events(timeout=2.0)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/connection.py", line 275, in drain_events
    return self.transport.drain_events(self.connection, **kwargs)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/__init__.py", line 830, in drain_events
    item, channel = get(timeout=timeout)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/scheduling.py", line 39, in get
    return self.fun(resource, **kwargs), resource
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/__init__.py", line 850, in _drain_channel
    return channel.drain_events(timeout=timeout)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/__init__.py", line 642, in drain_events
    return self._poll(self.cycle, timeout=timeout)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/__init__.py", line 328, in _poll
    return cycle.get()
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/scheduling.py", line 39, in get
    return self.fun(resource, **kwargs), resource
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/django/__init__.py", line 49, in _get
    m = Queue.objects.fetch(queue)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/django/managers.py", line 37, in fetch
    return queue.messages.pop()
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/django/managers.py", line 21, in _commit
    return fun(*args, **kwargs)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/django/managers.py", line 72, in pop
    result = resultset[0:1].get()
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/query.py", line 351, in get
    num = len(clone)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/query.py", line 122, in __len__
    self._fetch_all()
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/query.py", line 966, in _fetch_all
    self._result_cache = list(self.iterator())
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/query.py", line 265, in iterator
    for row in compiler.results_iter():
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/sql/compiler.py", line 700, in results_iter
    for rows in self.execute_sql(MULTI):
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/sql/compiler.py", line 786, in execute_sql
    cursor.execute(sql, params)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/utils.py", line 81, in execute
    return super(CursorDebugWrapper, self).execute(sql, params)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/utils.py", line 65, in execute
    return self.cursor.execute(sql, params)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/utils.py", line 94, in __exit__
    six.reraise(dj_exc_type, dj_exc_value, traceback)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/utils/six.py", line 658, in reraise
    raise value.with_traceback(tb)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/utils.py", line 65, in execute
    return self.cursor.execute(sql, params)
  File "/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/oracle/base.py", line 916, in execute
    return self.cursor.execute(query, self._param_generator(params))
django.db.utils.DatabaseError: ORA-00907: missing right parenthesis
```

This is discussed in google groups : https://groups.google.com/forum/#!topic/celery-users/5NOsMsw_p8A


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `kombu/transport/django/managers.py`
Content:
```
1 from __future__ import absolute_import
2 
3 from functools import wraps
4 
5 from django.db import transaction, connection, models
6 try:
7     from django.db import connections, router
8 except ImportError:  # pre-Django 1.2
9     connections = router = None  # noqa
10 
11 
12 try:
13     transaction.atomic
14 except AttributeError:
15     commit_on_success = transaction.commit_on_success
16 else:
17     def commit_on_success(fun):
18         @wraps(fun)
19         def _commit(*args, **kwargs):
20             with transaction.atomic():
21                 return fun(*args, **kwargs)
22         return _commit
23 
24 
25 
26 class QueueManager(models.Manager):
27 
28     def publish(self, queue_name, payload):
29         queue, created = self.get_or_create(name=queue_name)
30         queue.messages.create(payload=payload)
31 
32     def fetch(self, queue_name):
33         try:
34             queue = self.get(name=queue_name)
35         except self.model.DoesNotExist:
36             return
37 
38         return queue.messages.pop()
39 
40     def size(self, queue_name):
41         return self.get(name=queue_name).messages.count()
42 
43     def purge(self, queue_name):
44         try:
45             queue = self.get(name=queue_name)
46         except self.model.DoesNotExist:
47             return
48 
49         messages = queue.messages.all()
50         count = messages.count()
51         messages.delete()
52         return count
53 
54 
55 def select_for_update(qs):
56     try:
57         return qs.select_for_update()
58     except AttributeError:
59         return qs
60 
61 
62 class MessageManager(models.Manager):
63     _messages_received = [0]
64     cleanup_every = 10
65 
66     @commit_on_success
67     def pop(self):
68         try:
69             resultset = select_for_update(
70                 self.filter(visible=True).order_by('sent_at', 'id')
71             )
72             result = resultset[0:1].get()
73             result.visible = False
74             result.save()
75             recv = self.__class__._messages_received
76             recv[0] += 1
77             if not recv[0] % self.cleanup_every:
78                 self.cleanup()
79             return result.payload
80         except self.model.DoesNotExist:
81             pass
82 
83     def cleanup(self):
84         cursor = self.connection_for_write().cursor()
85         cursor.execute(
86             'DELETE FROM %s WHERE visible=%%s' % (
87                 self.model._meta.db_table, ),
88             (False, )
89         )
90 
91     def connection_for_write(self):
92         if connections:
93             return connections[router.db_for_write(self.model)]
94         return connection
95
```
--- 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/kombu/transport/django/managers.py b/kombu/transport/django/managers.py
--- a/kombu/transport/django/managers.py
+++ b/kombu/transport/django/managers.py
@@ -53,6 +53,8 @@
 
 
 def select_for_update(qs):
+    if connection.vendor == 'oracle':
+        return qs
     try:
         return qs.select_for_update()
     except AttributeError:

{"golden_diff": "diff --git a/kombu/transport/django/managers.py b/kombu/transport/django/managers.py\n--- a/kombu/transport/django/managers.py\n+++ b/kombu/transport/django/managers.py\n@@ -53,6 +53,8 @@\n \n \n def select_for_update(qs):\n+    if connection.vendor == 'oracle':\n+        return qs\n     try:\n         return qs.select_for_update()\n     except AttributeError:\n", "issue": "Oracle database backend throws exception\nWhen celery is used with oracle(11g) as database backend. There is an exception thrown\n\n```\n[2015-04-29 06:48:36,608: DEBUG/MainProcess] (0.011) QUERY = 'SELECT * FROM (SELECT ROWNUM AS \"_RN\", \"_SUB\".* FROM (SELECT \"DJKOMBU_MESSAGE\".\"ID\", \"DJKOMBU_MESSAGE\".\"VISIBLE\", \"DJKOMBU_MESSAGE\".\"SENT_AT\", \"DJKOMBU_MESSAGE\".\"PAYLOAD\", \"DJKOMBU_MESSAGE\".\"QUEUE_ID\" FROM \"DJKOMBU_MESSAGE\" WHERE (\"DJKOMBU_MESSAGE\".\"QUEUE_ID\" = :arg0 AND \"DJKOMBU_MESSAGE\".\"VISIBLE\" = :arg1) ORDER BY \"DJKOMBU_MESSAGE\".\"SENT_AT\" ASC, \"DJKOMBU_MESSAGE\".\"ID\" ASC FOR UPDATE) \"_SUB\" WHERE ROWNUM <= 1) WHERE \"_RN\" > 0' - PARAMS = (1, True); args=(1, True)\n[2015-04-29 06:48:36,617: ERROR/MainProcess] Unrecoverable error: DatabaseError(<cx_Oracle._Error object at 0x111105878>,)\nTraceback (most recent call last):\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/utils.py\", line 65, in execute\n    return self.cursor.execute(sql, params)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/oracle/base.py\", line 916, in execute\n    return self.cursor.execute(query, self._param_generator(params))\ncx_Oracle.DatabaseError: ORA-00907: missing right parenthesis\n\n\nThe above exception was the direct cause of the following exception:\n\nTraceback (most recent call last):\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/worker/__init__.py\", line 206, in start\n    self.blueprint.start(self)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/bootsteps.py\", line 123, in start\n    step.start(parent)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/bootsteps.py\", line 374, in start\n    return self.obj.start()\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/worker/consumer.py\", line 278, in start\n    blueprint.start(self)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/bootsteps.py\", line 123, in start\n    step.start(parent)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/worker/consumer.py\", line 821, in start\n    c.loop(*c.loop_args())\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/celery/worker/loops.py\", line 97, in synloop\n    connection.drain_events(timeout=2.0)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/connection.py\", line 275, in drain_events\n    return self.transport.drain_events(self.connection, **kwargs)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/__init__.py\", line 830, in drain_events\n    item, channel = get(timeout=timeout)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/scheduling.py\", line 39, in get\n    return self.fun(resource, **kwargs), resource\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/__init__.py\", line 850, in _drain_channel\n    return channel.drain_events(timeout=timeout)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/__init__.py\", line 642, in drain_events\n    return self._poll(self.cycle, timeout=timeout)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/__init__.py\", line 328, in _poll\n    return cycle.get()\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/virtual/scheduling.py\", line 39, in get\n    return self.fun(resource, **kwargs), resource\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/django/__init__.py\", line 49, in _get\n    m = Queue.objects.fetch(queue)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/django/managers.py\", line 37, in fetch\n    return queue.messages.pop()\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/django/managers.py\", line 21, in _commit\n    return fun(*args, **kwargs)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/kombu/transport/django/managers.py\", line 72, in pop\n    result = resultset[0:1].get()\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/query.py\", line 351, in get\n    num = len(clone)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/query.py\", line 122, in __len__\n    self._fetch_all()\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/query.py\", line 966, in _fetch_all\n    self._result_cache = list(self.iterator())\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/query.py\", line 265, in iterator\n    for row in compiler.results_iter():\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/sql/compiler.py\", line 700, in results_iter\n    for rows in self.execute_sql(MULTI):\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/models/sql/compiler.py\", line 786, in execute_sql\n    cursor.execute(sql, params)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/utils.py\", line 81, in execute\n    return super(CursorDebugWrapper, self).execute(sql, params)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/utils.py\", line 65, in execute\n    return self.cursor.execute(sql, params)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/utils.py\", line 94, in __exit__\n    six.reraise(dj_exc_type, dj_exc_value, traceback)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/utils/six.py\", line 658, in reraise\n    raise value.with_traceback(tb)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/utils.py\", line 65, in execute\n    return self.cursor.execute(sql, params)\n  File \"/Users/deepakn/dev/myapp/.env/lib/python3.4/site-packages/django/db/backends/oracle/base.py\", line 916, in execute\n    return self.cursor.execute(query, self._param_generator(params))\ndjango.db.utils.DatabaseError: ORA-00907: missing right parenthesis\n```\n\nThis is discussed in google groups : https://groups.google.com/forum/#!topic/celery-users/5NOsMsw_p8A\n\n", "before_files": [{"content": "from __future__ import absolute_import\n\nfrom functools import wraps\n\nfrom django.db import transaction, connection, models\ntry:\n    from django.db import connections, router\nexcept ImportError:  # pre-Django 1.2\n    connections = router = None  # noqa\n\n\ntry:\n    transaction.atomic\nexcept AttributeError:\n    commit_on_success = transaction.commit_on_success\nelse:\n    def commit_on_success(fun):\n        @wraps(fun)\n        def _commit(*args, **kwargs):\n            with transaction.atomic():\n                return fun(*args, **kwargs)\n        return _commit\n\n\n\nclass QueueManager(models.Manager):\n\n    def publish(self, queue_name, payload):\n        queue, created = self.get_or_create(name=queue_name)\n        queue.messages.create(payload=payload)\n\n    def fetch(self, queue_name):\n        try:\n            queue = self.get(name=queue_name)\n        except self.model.DoesNotExist:\n            return\n\n        return queue.messages.pop()\n\n    def size(self, queue_name):\n        return self.get(name=queue_name).messages.count()\n\n    def purge(self, queue_name):\n        try:\n            queue = self.get(name=queue_name)\n        except self.model.DoesNotExist:\n            return\n\n        messages = queue.messages.all()\n        count = messages.count()\n        messages.delete()\n        return count\n\n\ndef select_for_update(qs):\n    try:\n        return qs.select_for_update()\n    except AttributeError:\n        return qs\n\n\nclass MessageManager(models.Manager):\n    _messages_received = [0]\n    cleanup_every = 10\n\n    @commit_on_success\n    def pop(self):\n        try:\n            resultset = select_for_update(\n                self.filter(visible=True).order_by('sent_at', 'id')\n            )\n            result = resultset[0:1].get()\n            result.visible = False\n            result.save()\n            recv = self.__class__._messages_received\n            recv[0] += 1\n            if not recv[0] % self.cleanup_every:\n                self.cleanup()\n            return result.payload\n        except self.model.DoesNotExist:\n            pass\n\n    def cleanup(self):\n        cursor = self.connection_for_write().cursor()\n        cursor.execute(\n            'DELETE FROM %s WHERE visible=%%s' % (\n                self.model._meta.db_table, ),\n            (False, )\n        )\n\n    def connection_for_write(self):\n        if connections:\n            return connections[router.db_for_write(self.model)]\n        return connection\n", "path": "kombu/transport/django/managers.py"}], "after_files": [{"content": "from __future__ import absolute_import\n\nfrom functools import wraps\n\nfrom django.db import transaction, connection, models\ntry:\n    from django.db import connections, router\nexcept ImportError:  # pre-Django 1.2\n    connections = router = None  # noqa\n\n\ntry:\n    transaction.atomic\nexcept AttributeError:\n    commit_on_success = transaction.commit_on_success\nelse:\n    def commit_on_success(fun):\n        @wraps(fun)\n        def _commit(*args, **kwargs):\n            with transaction.atomic():\n                return fun(*args, **kwargs)\n        return _commit\n\n\n\nclass QueueManager(models.Manager):\n\n    def publish(self, queue_name, payload):\n        queue, created = self.get_or_create(name=queue_name)\n        queue.messages.create(payload=payload)\n\n    def fetch(self, queue_name):\n        try:\n            queue = self.get(name=queue_name)\n        except self.model.DoesNotExist:\n            return\n\n        return queue.messages.pop()\n\n    def size(self, queue_name):\n        return self.get(name=queue_name).messages.count()\n\n    def purge(self, queue_name):\n        try:\n            queue = self.get(name=queue_name)\n        except self.model.DoesNotExist:\n            return\n\n        messages = queue.messages.all()\n        count = messages.count()\n        messages.delete()\n        return count\n\n\ndef select_for_update(qs):\n    if connection.vendor == 'oracle':\n        return qs\n    try:\n        return qs.select_for_update()\n    except AttributeError:\n        return qs\n\n\nclass MessageManager(models.Manager):\n    _messages_received = [0]\n    cleanup_every = 10\n\n    @commit_on_success\n    def pop(self):\n        try:\n            resultset = select_for_update(\n                self.filter(visible=True).order_by('sent_at', 'id')\n            )\n            result = resultset[0:1].get()\n            result.visible = False\n            result.save()\n            recv = self.__class__._messages_received\n            recv[0] += 1\n            if not recv[0] % self.cleanup_every:\n                self.cleanup()\n            return result.payload\n        except self.model.DoesNotExist:\n            pass\n\n    def cleanup(self):\n        cursor = self.connection_for_write().cursor()\n        cursor.execute(\n            'DELETE FROM %s WHERE visible=%%s' % (\n                self.model._meta.db_table, ),\n            (False, )\n        )\n\n    def connection_for_write(self):\n        if connections:\n            return connections[router.db_for_write(self.model)]\n        return connection\n", "path": "kombu/transport/django/managers.py"}]}