diff --git a/mistral/db/v2/sqlalchemy/api.py b/mistral/db/v2/sqlalchemy/api.py
index 5894d6d0..d19c37c2 100644
--- a/mistral/db/v2/sqlalchemy/api.py
+++ b/mistral/db/v2/sqlalchemy/api.py
@@ -343,9 +343,26 @@ def delete_workflow_definition(name, session=None):
             "Workflow not found [workflow_name=%s]" % name
         )
 
+    cron_triggers = _get_associated_cron_triggers(name)
+
+    if cron_triggers:
+        raise exc.DBException(
+            "Can't delete workflow that has triggers [workflow_name=%s],"
+            "[cron_trigger_name(s)=%s]" % (name, ', '.join(cron_triggers))
+        )
+
     session.delete(wf_def)
 
 
+def _get_associated_cron_triggers(wf_name):
+    cron_triggers = _secure_query(
+        models.CronTrigger,
+        models.CronTrigger.name
+    ).filter_by(workflow_name=wf_name).all()
+
+    return [t[0] for t in cron_triggers]
+
+
 @b.session_aware()
 def delete_workflow_definitions(**kwargs):
     return _delete_all(models.WorkflowDefinition, **kwargs)
diff --git a/mistral/tests/functional/api/v2/test_mistral_basic_v2.py b/mistral/tests/functional/api/v2/test_mistral_basic_v2.py
index 1a21c4bf..5093a14d 100644
--- a/mistral/tests/functional/api/v2/test_mistral_basic_v2.py
+++ b/mistral/tests/functional/api/v2/test_mistral_basic_v2.py
@@ -330,6 +330,21 @@ class WorkflowTestsV2(base.TestCase):
                           self.client.update_request,
                           'workflows', 'wb_v1.yaml')
 
+    @test.attr(type='negative')
+    def test_delete_wf_with_trigger_associate(self):
+        tr_name = 'trigger'
+        resp, body = self.client.create_workflow('wf_v2.yaml')
+        name = body['workflows'][0]['name']
+        resp, body = self.client.create_cron_trigger(
+            tr_name, name, None, '5 * * * *')
+
+        self.assertRaises(exceptions.BadRequest,
+                          self.client.delete_obj,
+                          'workflows', name)
+
+        self.client.delete_obj('cron_triggers', tr_name)
+        self.client.triggers.remove(tr_name)
+
     @test.attr(type='negative')
     def test_delete_nonexistent_wf(self):
         self.assertRaises(exceptions.NotFound,