diff --git a/glance/db/sqlalchemy/api.py b/glance/db/sqlalchemy/api.py
index a80608434c..cf3f4e82b7 100644
--- a/glance/db/sqlalchemy/api.py
+++ b/glance/db/sqlalchemy/api.py
@@ -21,6 +21,7 @@
 """Defines interface for DB access."""
 
 from oslo.config import cfg
+from retrying import retry
 import six
 from six.moves import xrange
 import sqlalchemy
@@ -52,6 +53,15 @@ CONF.import_opt('connection', 'glance.openstack.common.db.options',
 _FACADE = None
 
 
+def _retry_on_deadlock(exc):
+    """Decorator to retry a DB API call if Deadlock was received."""
+
+    if isinstance(exc, db_exception.DBDeadlock):
+        LOG.warn(_("Deadlock detected. Retrying..."))
+        return True
+    return False
+
+
 def _create_facade_lazily():
     global _FACADE
     if _FACADE is None:
@@ -632,6 +642,8 @@ def _update_values(image_ref, values):
             setattr(image_ref, k, values[k])
 
 
+@retry(retry_on_exception=_retry_on_deadlock, wait_fixed=500,
+       stop_max_attempt_number=50)
 def _image_update(context, values, image_id, purge_props=False,
                   from_state=None):
     """
diff --git a/glance/tests/unit/test_db.py b/glance/tests/unit/test_db.py
index 60cfb6cfdd..ebcb0bb145 100644
--- a/glance/tests/unit/test_db.py
+++ b/glance/tests/unit/test_db.py
@@ -24,6 +24,8 @@ from glance.common import exception
 from glance.common import utils
 import glance.context
 import glance.db
+from glance.db.sqlalchemy import api
+from glance.openstack.common.db import exception as db_exc
 import glance.tests.unit.utils as unit_test_utils
 import glance.tests.utils as test_utils
 
@@ -689,3 +691,29 @@ class TestTaskRepo(test_utils.BaseTestCase):
         self.assertRaises(exception.NotFound,
                           self.task_repo.get,
                           task.task_id)
+
+
+class RetryOnDeadlockTestCase(test_utils.BaseTestCase):
+
+    def test_raise_deadlock(self):
+
+        class TestException(Exception):
+            pass
+
+        self.attempts = 3
+
+        def _mock_get_session():
+            def _raise_exceptions():
+                self.attempts -= 1
+                if self.attempts <= 0:
+                    raise TestException("Exit")
+                raise db_exc.DBDeadlock("Fake Exception")
+            return _raise_exceptions
+
+        with mock.patch.object(api, 'get_session') as sess:
+            sess.side_effect = _mock_get_session()
+
+            try:
+                api._image_update(None, {}, 'fake-id')
+            except TestException:
+                self.assertEqual(sess.call_count, 3)