diff --git a/glance/db/sqlalchemy/api.py b/glance/db/sqlalchemy/api.py
index 9cddd2bfc6..28fe3cc390 100644
--- a/glance/db/sqlalchemy/api.py
+++ b/glance/db/sqlalchemy/api.py
@@ -1322,6 +1322,35 @@ def purge_deleted_rows(context, age_in_days, max_rows, session=None):
             continue
         if hasattr(model_class, 'deleted'):
             tables.append(model_class.__tablename__)
+
+    # First force purging of records that are not soft deleted but
+    # are referencing soft deleted tasks/images records (e.g. task_info
+    # records). Then purge all soft deleted records in glance tables in the
+    # right order to avoid FK constraint violation.
+    t = Table("tasks", metadata, autoload=True)
+    ti = Table("task_info", metadata, autoload=True)
+    joined_rec = ti.join(t, t.c.id == ti.c.task_id)
+    deleted_task_info = sql.select([ti.c.task_id],
+                                   t.c.deleted_at < deleted_age).\
+        select_from(joined_rec).order_by(t.c.deleted_at).limit(max_rows)
+    delete_statement = DeleteFromSelect(ti, deleted_task_info,
+                                        ti.c.task_id)
+    LOG.info(_LI('Purging deleted rows older than %(age_in_days)d day(s) '
+                 'from table %(tbl)s'),
+             {'age_in_days': age_in_days, 'tbl': ti})
+    try:
+        with session.begin():
+            result = session.execute(delete_statement)
+    except (db_exception.DBError, db_exception.DBReferenceError) as ex:
+        LOG.exception(_LE('DBError detected when force purging '
+                          'table=%(table)s: %(error)s'),
+                      {'table': ti, 'error': six.text_type(ex)})
+        raise
+
+    rows = result.rowcount
+    LOG.info(_LI('Deleted %(rows)d row(s) from table %(tbl)s'),
+             {'rows': rows, 'tbl': ti})
+
     # get rid of FK constraints
     for tbl in ('images', 'tasks'):
         try:
diff --git a/glance/tests/functional/db/base.py b/glance/tests/functional/db/base.py
index 0b08f83702..6d8a4dd892 100644
--- a/glance/tests/functional/db/base.py
+++ b/glance/tests/functional/db/base.py
@@ -1946,6 +1946,7 @@ class DBPurgeTests(test_utils.BaseTestCase):
         self.adm_context = context.get_admin_context(show_deleted=True)
         self.db_api = db_tests.get_db(self.config)
         db_tests.reset_db(self.db_api)
+        self.context = context.RequestContext(is_admin=True)
         self.image_fixtures, self.task_fixtures = self.build_fixtures()
         self.create_tasks(self.task_fixtures)
         self.create_images(self.image_fixtures)
@@ -2079,6 +2080,29 @@ class DBPurgeTests(test_utils.BaseTestCase):
         images_rows = session.query(images).count()
         self.assertEqual(4, images_rows)
 
+    def test_purge_task_info_with_refs_to_soft_deleted_tasks(self):
+        session = db_api.get_session()
+        engine = db_api.get_engine()
+
+        # check initial task and task_info row number are 3
+        tasks = self.db_api.task_get_all(self.adm_context)
+        self.assertEqual(3, len(tasks))
+
+        task_info = sqlalchemyutils.get_table(engine, 'task_info')
+        task_info_rows = session.query(task_info).count()
+        self.assertEqual(3, task_info_rows)
+
+        # purge soft deleted rows older than yesterday
+        self.db_api.purge_deleted_rows(self.context, 1, 5)
+
+        # check 1 row of task table is purged
+        tasks = self.db_api.task_get_all(self.adm_context)
+        self.assertEqual(2, len(tasks))
+
+        # and no task_info was left behind, 1 row purged
+        task_info_rows = session.query(task_info).count()
+        self.assertEqual(2, task_info_rows)
+
 
 class TestVisibility(test_utils.BaseTestCase):
     def setUp(self):