From de804d7245dd203bc63e4493162bcdf5e8646440 Mon Sep 17 00:00:00 2001 From: Denis Kataev Date: Mon, 12 Mar 2018 11:40:34 -0400 Subject: 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 --- test/dialect/test_sqlite.py | 136 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 135 insertions(+), 1 deletion(-) (limited to 'test/dialect/test_sqlite.py') 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): -- cgit v1.2.1