Implement SQLite ON CONFLICT for constraints

Implemented the SQLite ``ON CONFLICT`` clause as understood at the DDL
level, e.g. for primary key, unique, and CHECK constraints as well as
specified on a :class:`.Column` to satisfy inline primary key and NOT NULL.
Pull request courtesy Denis Kataev.

Fixes: #4360
Change-Id: I4cd4bafa8fca41e3101c87dbbfe169741bbda3f4
Pull-request: https://github.com/zzzeek/sqlalchemy/pull/431

1 files changed, 135 insertions, 1 deletions

diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index 9b1d5d7ea..417ace5c8 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -8,7 +8,8 @@ from sqlalchemy.testing import eq_, assert_raises, \
     assert_raises_message, is_, expect_warnings
 from sqlalchemy import Table, select, bindparam, Column,\
     MetaData, func, extract, ForeignKey, text, DefaultClause, and_, \
-    create_engine, UniqueConstraint, Index, PrimaryKeyConstraint
+    create_engine, \
+    UniqueConstraint, Index, PrimaryKeyConstraint, CheckConstraint
 from sqlalchemy.types import Integer, String, Boolean, DateTime, Date, Time
 from sqlalchemy import types as sqltypes
 from sqlalchemy import event, inspect
@@ -835,6 +836,139 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
             CreateTable(t).compile, dialect=sqlite.dialect()
         )
 
+class OnConflictDDLTest(fixtures.TestBase, AssertsCompiledSQL):
+
+    __dialect__ = sqlite.dialect()
+
+    def test_on_conflict_clause_column_not_null(self):
+        c = Column('test', Integer, nullable=False,
+                   sqlite_on_conflict_not_null='FAIL')
+
+        self.assert_compile(schema.CreateColumn(c),
+                            'test INTEGER NOT NULL '
+                            'ON CONFLICT FAIL', dialect=sqlite.dialect())
+
+    def test_on_conflict_clause_column_many_clause(self):
+        meta = MetaData()
+        t = Table(
+            'n', meta,
+            Column('test', Integer, nullable=False, primary_key=True,
+                   sqlite_on_conflict_not_null='FAIL',
+                   sqlite_on_conflict_primary_key='IGNORE')
+        )
+
+        self.assert_compile(CreateTable(t),
+                            "CREATE TABLE n ("
+                            "test INTEGER NOT NULL ON CONFLICT FAIL, "
+                            "PRIMARY KEY (test) ON CONFLICT IGNORE)",
+                            dialect=sqlite.dialect())
+
+    def test_on_conflict_clause_unique_constraint_from_column(self):
+        meta = MetaData()
+        t = Table(
+            'n', meta,
+            Column('x', String(30), unique=True,
+                   sqlite_on_conflict_unique='FAIL'),
+        )
+
+        self.assert_compile(CreateTable(t),
+                            "CREATE TABLE n (x VARCHAR(30), "
+                            "UNIQUE (x) ON CONFLICT FAIL)",
+                            dialect=sqlite.dialect())
+
+    def test_on_conflict_clause_unique_constraint(self):
+
+        meta = MetaData()
+        t = Table(
+            'n', meta,
+            Column('id', Integer),
+            Column('x', String(30)),
+            UniqueConstraint('id', 'x', sqlite_on_conflict='FAIL'),
+        )
+
+        self.assert_compile(CreateTable(t),
+                            "CREATE TABLE n (id INTEGER, x VARCHAR(30), "
+                            "UNIQUE (id, x) ON CONFLICT FAIL)",
+                            dialect=sqlite.dialect())
+
+    def test_on_conflict_clause_primary_key(self):
+
+        meta = MetaData()
+        t = Table(
+            'n', meta,
+            Column('id', Integer, primary_key=True,
+                   sqlite_on_conflict_primary_key='FAIL'),
+            sqlite_autoincrement=True
+        )
+
+        self.assert_compile(CreateTable(t),
+                            "CREATE TABLE n (id INTEGER NOT NULL "
+                            "PRIMARY KEY ON CONFLICT FAIL AUTOINCREMENT)",
+                            dialect=sqlite.dialect())
+
+    def test_on_conflict_clause_primary_key_constraint_from_column(self):
+
+        meta = MetaData()
+        t = Table(
+            'n', meta,
+            Column('x', String(30), sqlite_on_conflict_primary_key='FAIL',
+                   primary_key=True),
+        )
+
+        self.assert_compile(CreateTable(t),
+                            "CREATE TABLE n (x VARCHAR(30) NOT NULL, "
+                            "PRIMARY KEY (x) ON CONFLICT FAIL)",
+                            dialect=sqlite.dialect())
+
+    def test_on_conflict_clause_check_constraint(self):
+
+        meta = MetaData()
+        t = Table(
+            'n', meta,
+            Column('id', Integer),
+            Column('x', Integer),
+            CheckConstraint('id > x', sqlite_on_conflict='FAIL'),
+        )
+
+        self.assert_compile(CreateTable(t),
+                            "CREATE TABLE n (id INTEGER, x INTEGER, "
+                            "CHECK (id > x) ON CONFLICT FAIL)",
+                            dialect=sqlite.dialect())
+
+    def test_on_conflict_clause_check_constraint_from_column(self):
+
+        meta = MetaData()
+        t = Table(
+            'n', meta,
+            Column('x', Integer,
+                   CheckConstraint('x > 1',
+                                   sqlite_on_conflict='FAIL')),
+        )
+
+        assert_raises_message(
+            exc.CompileError,
+            "SQLite does not support on conflict "
+            "clause for column check constraint",
+            CreateTable(t).compile, dialect=sqlite.dialect()
+        )
+
+    def test_on_conflict_clause_primary_key_constraint(self):
+
+        meta = MetaData()
+        t = Table(
+            'n', meta,
+            Column('id', Integer),
+            Column('x', String(30)),
+            PrimaryKeyConstraint('id', 'x', sqlite_on_conflict='FAIL'),
+        )
+
+        self.assert_compile(CreateTable(t),
+                            "CREATE TABLE n ("
+                            "id INTEGER NOT NULL, "
+                            "x VARCHAR(30) NOT NULL, "
+                            "PRIMARY KEY (id, x) ON CONFLICT FAIL)",
+                            dialect=sqlite.dialect())
+
 
 class InsertTest(fixtures.TestBase, AssertsExecutionResults):