Handle CHECK constraint integrity in PostgreSQL

PostgreSQL offers CHECK constraints integrity which are currently raised
as generic error. This patch add a filter so they can have their own
exception type.

Change-Id: I7e36ef4385b227d1e62e18277d470047a369a238

3 files changed, 50 insertions, 0 deletions

diff --git a/oslo_db/exception.py b/oslo_db/exception.py
index f950f6a..3b9c3f9 100644
--- a/oslo_db/exception.py
+++ b/oslo_db/exception.py
@@ -87,6 +87,23 @@ class DBDuplicateEntry(DBError):
         super(DBDuplicateEntry, self).__init__(inner_exception)
 
 
+class DBConstraintError(DBError):
+    """Check constraint fails for column error.
+
+    Raised when made an attempt to write to a column a value that does not
+    satisfy a CHECK constraint.
+
+    :kwarg table: the table name for which the check fails
+    :type table: str
+    :kwarg check_name: the table of the check that failed to be satisfied
+    :type check_name: str
+    """
+    def __init__(self, table, check_name, inner_exception=None):
+        self.table = table
+        self.check_name = check_name
+        super(DBConstraintError, self).__init__(inner_exception)
+
+
 class DBReferenceError(DBError):
     """Foreign key violation error.
 
diff --git a/oslo_db/sqlalchemy/exc_filters.py b/oslo_db/sqlalchemy/exc_filters.py
index 777fda6..8da503d 100644
--- a/oslo_db/sqlalchemy/exc_filters.py
+++ b/oslo_db/sqlalchemy/exc_filters.py
@@ -220,6 +220,26 @@ def _foreign_key_error(integrity_error, match, engine_name, is_disconnect):
                                      integrity_error)
 
 
+@filters("postgresql", sqla_exc.IntegrityError,
+         r".*new row for relation \"(?P<table>.+)\" "
+         "violates check constraint "
+         "\"(?P<check_name>.+)\"")
+def _check_constraint_error(
+        integrity_error, match, engine_name, is_disconnect):
+    """Filter for check constraint errors."""
+
+    try:
+        table = match.group("table")
+    except IndexError:
+        table = None
+    try:
+        check_name = match.group("check_name")
+    except IndexError:
+        check_name = None
+
+    raise exception.DBConstraintError(table, check_name, integrity_error)
+
+
 @filters("ibm_db_sa", sqla_exc.IntegrityError, r"^.*SQL0803N.*$")
 def _db2_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
     """Filter for DB2 duplicate key errors.
diff --git a/oslo_db/tests/sqlalchemy/test_exc_filters.py b/oslo_db/tests/sqlalchemy/test_exc_filters.py
index aafdcfb..1a35fba 100644
--- a/oslo_db/tests/sqlalchemy/test_exc_filters.py
+++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py
@@ -354,6 +354,19 @@ class TestReferenceErrorMySQL(TestReferenceErrorSQLite,
         self.assertEqual("resource_foo", matched.key_table)
 
 
+class TestConstraint(TestsExceptionFilter):
+    def test_postgresql(self):
+        matched = self._run_test(
+            "postgresql", "insert into resource some_values",
+            self.IntegrityError(
+                "new row for relation \"resource\" violates "
+                "check constraint \"ck_started_before_ended\""),
+            exception.DBConstraintError,
+        )
+        self.assertEqual("resource", matched.table)
+        self.assertEqual("ck_started_before_ended", matched.check_name)
+
+
 class TestDuplicate(TestsExceptionFilter):
 
     def _run_dupe_constraint_test(self, dialect_name, message,