:ticket:`3850`
+.. _change_4360:
+
+Support for SQLite ON CONFLICT in constraints added
+----------------------------------------------------
+
+SQLite supports a non-standard ON CONFLICT clause that may be specified
+for standalone constraints as well as some column-inline constraints such as
+NOT NULL. Support has been added for these clauses via the ``sqlite_on_conflict``
+keyword added to objects like :class:`.UniqueConstraint` as well
+as several :class:`.Column` -specific variants::
+
+ some_table = Table(
+ 'some_table', metadata,
+ Column('id', Integer, primary_key=True, sqlite_on_conflict_primary_key='FAIL'),
+ Column('data', Integer),
+ UniqueConstraint('id', 'data', sqlite_on_conflict='IGNORE')
+ )
+
+The above table would render in a CREATE TABLE statement as::
+
+ CREATE TABLE some_table (
+ id INTEGER NOT NULL,
+ data INTEGER,
+ PRIMARY KEY (id) ON CONFLICT FAIL,
+ UNIQUE (id, data) ON CONFLICT IGNORE
+ )
+
+.. seealso::
+
+ :ref:`sqlite_on_conflict_ddl`
+
+:ticket:`4360`
Dialect Improvements and Changes - Oracle
=============================================
--- /dev/null
+.. change::
+ :tags: feature, sqlite
+ :tickets: 4360
+
+ 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.
+
+ .. seealso::
+
+ :ref:`change_4360`
:ref:`use_alter` - more information on SQLAlchemy's facilities for handling
mutually-dependent foreign key constraints.
+.. _sqlite_on_conflict_ddl:
+
+ON CONFLICT support for constraints
+-----------------------------------
+
+SQLite supports a non-standard clause known as ON CONFLICT which can be applied
+to primary key, unique, check, and not null constraints. In DDL, it is
+rendered either within the "CONSTRAINT" clause or within the column definition
+itself depending on the location of the target constraint. To render this
+clause within DDL, the extension parameter ``sqlite_on_conflict`` can be
+specified with a string conflict resolution algorithm within the
+:class:`.PrimaryKeyConstraint`, :class:`.UniqueConstraint`,
+:class:`.CheckConstraint` objects. Within the :class:`.Column` object, there
+are individual parameters ``sqlite_on_conflict_not_null``,
+``sqlite_on_conflict_primary_key``, ``sqlite_on_conflict_unique`` which each
+correspond to the three types of relevant constraint types that can be
+indicated from a :class:`.Column` object.
+
+.. seealso::
+
+ `ON CONFLICT <https://www.sqlite.org/lang_conflict.html>`_ - in the SQLite
+ documentation
+
+.. versionadded:: 1.3
+
+
+The ``sqlite_on_conflict`` parameters accept a string argument which is just
+the resolution name to be chosen, which on SQLite can be one of ROLLBACK,
+ABORT, FAIL, IGNORE, and REPLACE. For example, to add a UNIQUE constraint
+that specifies the IGNORE algorithm::
+
+ some_table = Table(
+ 'some_table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', Integer),
+ UniqueConstraint('id', 'data', sqlite_on_conflict='IGNORE')
+ )
+
+The above renders CREATE TABLE DDL as::
+
+ CREATE TABLE some_table (
+ id INTEGER NOT NULL,
+ data INTEGER,
+ PRIMARY KEY (id),
+ UNIQUE (id, data) ON CONFLICT IGNORE
+ )
+
+
+When using the :paramref:`.Column.unique` flag to add a UNIQUE constraint
+to a single column, the ``sqlite_on_conflict_unique`` parameter can
+be added to the :class:`.Column` as well, which will be added to the
+UNIQUE constraint in the DDL::
+
+ some_table = Table(
+ 'some_table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', Integer, unique=True,
+ sqlite_on_conflict_unique='IGNORE')
+ )
+
+rendering::
+
+ CREATE TABLE some_table (
+ id INTEGER NOT NULL,
+ data INTEGER,
+ PRIMARY KEY (id),
+ UNIQUE (data) ON CONFLICT IGNORE
+ )
+
+To apply the FAIL algorithm for a NOT NULL constraint,
+``sqlite_on_conflict_not_null`` is used::
+
+ some_table = Table(
+ 'some_table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', Integer, nullable=False,
+ sqlite_on_conflict_not_null='FAIL')
+ )
+
+this renders the column inline ON CONFLICT phrase::
+
+ CREATE TABLE some_table (
+ id INTEGER NOT NULL,
+ data INTEGER NOT NULL ON CONFLICT FAIL,
+ PRIMARY KEY (id)
+ )
+
+
+Similarly, for an inline primary key, use ``sqlite_on_conflict_primary_key``::
+
+ some_table = Table(
+ 'some_table', metadata,
+ Column('id', Integer, primary_key=True,
+ sqlite_on_conflict_primary_key='FAIL')
+ )
+
+SQLAlchemy renders the PRIMARY KEY constraint separately, so the conflict
+resolution algorithm is applied to the constraint itself::
+
+ CREATE TABLE some_table (
+ id INTEGER NOT NULL,
+ PRIMARY KEY (id) ON CONFLICT FAIL
+ )
+
.. _sqlite_type_reflection:
Type Reflection
if not column.nullable:
colspec += " NOT NULL"
+ on_conflict_clause = column.dialect_options['sqlite'][
+ 'on_conflict_not_null']
+ if on_conflict_clause is not None:
+ colspec += " ON CONFLICT " + on_conflict_clause
+
if column.primary_key:
if (
column.autoincrement is True and
if (column.table.dialect_options['sqlite']['autoincrement'] and
len(column.table.primary_key.columns) == 1 and
- issubclass(column.type._type_affinity, sqltypes.Integer) and
+ issubclass(
+ column.type._type_affinity, sqltypes.Integer) and
not column.foreign_keys):
- colspec += " PRIMARY KEY AUTOINCREMENT"
+ colspec += " PRIMARY KEY"
+
+ on_conflict_clause = column.dialect_options['sqlite'][
+ 'on_conflict_primary_key']
+ if on_conflict_clause is not None:
+ colspec += " ON CONFLICT " + on_conflict_clause
+
+ colspec += " AUTOINCREMENT"
return colspec
not c.foreign_keys):
return None
- return super(SQLiteDDLCompiler, self).visit_primary_key_constraint(
- constraint)
+ text = super(
+ SQLiteDDLCompiler,
+ self).visit_primary_key_constraint(constraint)
+
+ on_conflict_clause = constraint.dialect_options['sqlite'][
+ 'on_conflict']
+ if on_conflict_clause is None and len(constraint.columns) == 1:
+ on_conflict_clause = list(constraint)[0].\
+ dialect_options['sqlite']['on_conflict_primary_key']
+
+ if on_conflict_clause is not None:
+ text += " ON CONFLICT " + on_conflict_clause
+
+ return text
+
+ def visit_unique_constraint(self, constraint):
+ text = super(
+ SQLiteDDLCompiler,
+ self).visit_unique_constraint(constraint)
+
+ on_conflict_clause = constraint.dialect_options['sqlite'][
+ 'on_conflict']
+ if on_conflict_clause is None and len(constraint.columns) == 1:
+ on_conflict_clause = list(constraint)[0].\
+ dialect_options['sqlite']['on_conflict_unique']
+
+ if on_conflict_clause is not None:
+ text += " ON CONFLICT " + on_conflict_clause
+
+ return text
+
+ def visit_check_constraint(self, constraint):
+ text = super(
+ SQLiteDDLCompiler,
+ self).visit_check_constraint(constraint)
+
+ on_conflict_clause = constraint.dialect_options['sqlite'][
+ 'on_conflict']
+
+ if on_conflict_clause is not None:
+ text += " ON CONFLICT " + on_conflict_clause
+
+ return text
+
+ def visit_column_check_constraint(self, constraint):
+ text = super(
+ SQLiteDDLCompiler,
+ self).visit_column_check_constraint(constraint)
+
+ if constraint.dialect_options['sqlite']['on_conflict'] is not None:
+ raise exc.CompileError(
+ "SQLite does not support on conflict clause for "
+ "column check constraint")
+
+ return text
def visit_foreign_key_constraint(self, constraint):
(sa_schema.Index, {
"where": None,
}),
+ (sa_schema.Column, {
+ "on_conflict_primary_key": None,
+ "on_conflict_not_null": None,
+ "on_conflict_unique": None,
+ }),
+ (sa_schema.Constraint, {
+ "on_conflict": None,
+ }),
]
_broken_fk_pragma_quotes = False
return self._schema_item_copy(table)
-class Column(SchemaItem, ColumnClause):
+class Column(DialectKWArgs, SchemaItem, ColumnClause):
"""Represents a column in a database table."""
__visit_name__ = 'column'
if 'info' in kwargs:
self.info = kwargs.pop('info')
- if kwargs:
- raise exc.ArgumentError(
- "Unknown arguments passed to Column: " + repr(list(kwargs)))
+ self._extra_kwargs(**kwargs)
+
+ def _extra_kwargs(self, **kwargs):
+ self._validate_dialect_kwargs(kwargs)
# @property
# def quote(self):
def __init__(self, sqltext, name=None, deferrable=None,
initially=None, table=None, info=None, _create_rule=None,
- _autoattach=True, _type_bound=False):
+ _autoattach=True, _type_bound=False, **kw):
r"""Construct a CHECK constraint.
:param sqltext:
name=name, deferrable=deferrable,
initially=initially, _create_rule=_create_rule, info=info,
_type_bound=_type_bound, _autoattach=_autoattach,
- *columns)
+ *columns, **kw)
if table is not None:
self._set_parent_with_dispatch(table)
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
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):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
