--- /dev/null
+.. change::
+ :tags: schema, sql
+ :tickets: 5362, 5324, 5360
+
+ Added the :class:`_schema.Identity` construct that can be used to
+ configure identity columns rendered with GENERATED { ALWAYS |
+ BY DEFAULT } AS IDENTITY. Currently the supported backends are
+ PostgreSQL >= 10, Oracle >= 12 and MSSQL (with different syntax
+ and a subset of functionalities).
* Microsoft SQL Server
+* SQLite as of version 3.31
+
* Firebird
When :class:`.Computed` is used with an unsupported backend, if the target
:class:`.Computed`
+.. _identity_ddl:
+
+Identity Columns (GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY)
+--------------------------------------------------------------
+
+.. versionadded:: 1.4
+
+The :class:`.Identity` construct allows a :class:`_schema.Column` to be declared
+as an identity column and rendered in DDL as "GENERATED { ALWAYS | BY DEFAULT }
+AS IDENTITY". An identity column has its value automatically generated by the
+database server using an incrementing (or decrementing) sequence. The construct
+shares most of its option to control the database behaviour with
+:class:`.Sequence`.
+
+Example::
+
+ from sqlalchemy import Table, Column, MetaData, Integer, Computed
+
+ metadata = MetaData()
+
+ data = Table(
+ "data",
+ metadata,
+ Column('id', Integer, Identity(start=42, cycle=True), primary_key=True),
+ Column('data', String)
+ )
+
+The DDL for the ``data`` table when run on a PostgreSQL 12 backend will look
+like::
+
+ CREATE TABLE data (
+ id INTEGER GENERATED BY DEFAULT AS IDENTITY (START WITH 42 CYCLE) NOT NULL,
+ data VARCHAR,
+ PRIMARY KEY (id)
+ )
+
+The database will generate a value for the ``id`` column upon insert,
+starting from ``42``, if the statement did not already contain a value for
+the ``id`` column.
+An identity column can also require that the database generate the value
+of the column, ignoring the value passed with the statement or raising an
+error, depending on backend. To activate this mode, specify the parameter
+``always=True`` in the :class:`.Identity` construct. Updating the previous
+example to include this parameter will generate the following DDL::
+
+ CREATE TABLE data (
+ id INTEGER GENERATED ALWAYS AS IDENTITY (START WITH 42 CYCLE) NOT NULL,
+ data VARCHAR,
+ PRIMARY KEY (id)
+ )
+
+..TODO: not sure if we will actually implement it this way. server onupdate seems
+ unnecessary?
+
+The :class:`.Identity` construct is a subclass of the :class:`.FetchedValue`
+object, and will set itself up as the "server default" generator for the
+target :class:`_schema.Column`, meaning it will be treated
+as a default generating column when INSERT statements are generated,
+as well as that it will be fetched as a generating column when using the ORM.
+This includes that it will be part of the RETURNING clause of the database
+for databases which support RETURNING and the generated values are to be
+eagerly fetched.
+
+The :class:`.Identity` construct is currently known to be supported by:
+
+* PostgreSQL as of version 10.
+
+* Oracle as of version 12. Oracle also support passing ``always=None`` to
+ enable the default generated mode and the parameter ``on_null=True`` to
+ specify "ON NULL" in conjunction with a 'by default' identity column.
+
+* Microsoft SQL Server. MSSQL uses a custom syntax that only supports the
+ ``start`` and ``increment`` parameters, and ignores all other.
+
+When :class:`.Identity` is used with an unsupported backend, it is ignored,
+and the default SQLAlchemy logic for autoincrementing columns is used.
+
+An error is raised when a :class:`_schema.Column` specifies both an
+:class:`.Identity` and also set :paramref:`_schema.Column.autoincrement`
+to ``False``.
+
+.. seealso::
+
+ :class:`.Identity`
+
+
Default Objects API
-------------------
:members:
-.. autoclass:: IdentityOptions
+.. autoclass:: Identity
+ :members:
from .schema import FetchedValue # noqa
from .schema import ForeignKey # noqa
from .schema import ForeignKeyConstraint # noqa
-from .schema import IdentityOptions # noqa
+from .schema import Identity # noqa
from .schema import Index # noqa
from .schema import MetaData # noqa
from .schema import PrimaryKeyConstraint # noqa
.. sourcecode:: sql
CREATE TABLE t (
- id INTEGER NOT NULL IDENTITY(1,1),
+ id INTEGER NOT NULL IDENTITY,
x INTEGER NULL,
PRIMARY KEY (id)
)
Column('x', Integer, autoincrement=True))
m.create_all(engine)
-.. versionchanged:: 1.3 Added ``mssql_identity_start`` and
- ``mssql_identity_increment`` parameters to :class:`_schema.Column`.
- These replace
+.. versionchanged:: 1.4 Added :class:`_schema.Identity` construct
+ in a :class:`_schema.Column` to specify the start and increment
+ parameters of an IDENTITY. These replace
the use of the :class:`.Sequence` object in order to specify these values.
+.. deprecated:: 1.4
+
+ The ``mssql_identity_start`` and ``mssql_identity_increment`` parameters
+ to :class:`_schema.Column` are deprecated and should we replaced by
+ an :class:`_schema.Identity` object. Specifying both ways of configuring
+ an IDENTITY will result in a compile error.
+
.. deprecated:: 1.3
The use of :class:`.Sequence` to specify IDENTITY characteristics is
deprecated and will be removed in a future release. Please use
- the ``mssql_identity_start`` and ``mssql_identity_increment`` parameters
- documented at :ref:`mssql_identity`.
+ the :class:`_schema.Identity` object parameters
+ :paramref:`_schema.Identity.start` and
+ :paramref:`_schema.Identity.increment`.
.. versionchanged:: 1.4 Removed the ability to use a :class:`.Sequence`
object to modify IDENTITY characteristics. :class:`.Sequence` objects
Specific control over the "start" and "increment" values for
the ``IDENTITY`` generator are provided using the
-``mssql_identity_start`` and ``mssql_identity_increment`` parameters
-passed to the :class:`_schema.Column` object::
+:paramref:`_schema.Identity.start` and :paramref:`_schema.Identity.increment`
+parameters passed to the :class:`_schema.Identity` object::
- from sqlalchemy import Table, Integer, Column
+ from sqlalchemy import Table, Integer, Column, Identity
test = Table(
'test', metadata,
Column(
- 'id', Integer, primary_key=True, mssql_identity_start=100,
- mssql_identity_increment=10
+ 'id',
+ Integer,
+ primary_key=True,
+ Identity(start=100, increment=10)
),
Column('name', String(20))
)
name VARCHAR(20) NULL,
)
-.. versionchanged:: 1.3 The ``mssql_identity_start`` and
- ``mssql_identity_increment`` parameters are now used to affect the
+.. note::
+
+ The :class:`_schema.Identity` object supports many other parameter in
+ addition to ``start`` and ``increment``. These are not supported by
+ SQL Server and will be ignored when generating the CREATE TABLE ddl.
+
+.. versionchanged:: 1.3.19 The :class:`_schema.Identity` object is
+ now used to affect the
``IDENTITY`` generator for a :class:`_schema.Column` under SQL Server.
Previously, the :class:`.Sequence` object was used. As SQL Server now
supports real sequences as a separate construct, :class:`.Sequence` will be
- functional in the normal way in a future SQLAlchemy version.
+ functional in the normal way starting from SQLAlchemy version 1.4.
INSERT behavior
^^^^^^^^^^^^^^^^
from . import information_schema as ischema
from ... import exc
+from ... import Identity
from ... import schema as sa_schema
from ... import Sequence
from ... import sql
or column.primary_key
or isinstance(column.default, sa_schema.Sequence)
or column.autoincrement is True
+ or column.identity
):
colspec += " NOT NULL"
elif column.computed is None:
"in order to generate DDL"
)
- if (
+ d_opt = column.dialect_options["mssql"]
+ start = d_opt["identity_start"]
+ increment = d_opt["identity_increment"]
+ if start is not None or increment is not None:
+ if column.identity:
+ raise exc.CompileError(
+ "Cannot specify options 'mssql_identity_start' and/or "
+ "'mssql_identity_increment' while also using the "
+ "'Identity' construct."
+ )
+ util.warn_deprecated(
+ "The dialect options 'mssql_identity_start' and "
+ "'mssql_identity_increment' are deprecated. "
+ "Use the 'Identity' object instead.",
+ "1.4",
+ )
+
+ if column.identity:
+ colspec += self.process(column.identity, **kwargs)
+ elif (
column is column.table._autoincrement_column
or column.autoincrement is True
):
if not isinstance(column.default, Sequence):
- start = column.dialect_options["mssql"]["identity_start"]
- increment = column.dialect_options["mssql"][
- "identity_increment"
- ]
- colspec += " IDENTITY(%s,%s)" % (start, increment)
+ colspec += self.process(
+ Identity(start=start, increment=increment)
+ )
else:
default = self.get_column_default_string(column)
if default is not None:
create, prefix=prefix, **kw
)
+ def visit_identity_column(self, identity, **kw):
+ text = " IDENTITY"
+ if identity.start is not None or identity.increment is not None:
+ start = 1 if identity.start is None else identity.start
+ increment = 1 if identity.increment is None else identity.increment
+ text += "(%s,%s)" % (start, increment)
+ return text
+
class MSIdentifierPreparer(compiler.IdentifierPreparer):
reserved_words = RESERVED_WORDS
(sa_schema.PrimaryKeyConstraint, {"clustered": None}),
(sa_schema.UniqueConstraint, {"clustered": None}),
(sa_schema.Index, {"clustered": None, "include": None, "where": None}),
- (sa_schema.Column, {"identity_start": 1, "identity_increment": 1}),
+ (
+ sa_schema.Column,
+ {"identity_start": None, "identity_increment": None},
+ ),
]
def __init__(
SQLAlchemy Table objects which include integer primary keys are usually
assumed to have "autoincrementing" behavior, meaning they can generate their
-own primary key values upon INSERT. Since Oracle has no "autoincrement"
+own primary key values upon INSERT. For use within Oracle, two options are
+available, which are the use of IDENTITY columns (Oracle 12 and above only)
+or the association of a SEQUENCE with the column.
+
+Specifying GENERATED AS IDENTITY (Oracle 12 and above)
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Starting from version 12 Oracle can make use of identity columns using
+the :class:`_sql.Identity` to specify the autoincrementing behavior::
+
+ t = Table('mytable', metadata,
+ Column('id', Integer, Identity(start=3), primary_key=True),
+ Column(...), ...
+ )
+
+The CREATE TABLE for the above :class:`_schema.Table` object would be:
+
+.. sourcecode:: sql
+
+ CREATE TABLE mytable (
+ id INTEGER GENERATED BY DEFAULT AS IDENTITY (START WITH 3) NOT NULL,
+ ...,
+ PRIMARY KEY (id)
+ )
+
+The :class:`_schema.Identity` object support many options to control the
+"autoincrementing" behavior of the column, like the starting value, the
+incrementing value, etc.
+In addition to the standard options, Oracle supports setting
+:paramref:`_schema.Identity.always` to ``None`` to use the default
+generated mode, rendering GENERATED AS IDENTITY in the DDL. It also supports
+setting :paramref:`_schema.Identity.on_null` to ``True`` to specify ON NULL
+in conjunction with a 'BY DEFAULT' identity column.
+
+Using a SEQUENCE (all Oracle versions)
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Older version of Oracle had no "autoincrement"
feature, SQLAlchemy relies upon sequences to produce these values. With the
-Oracle dialect, *a sequence must always be explicitly specified to enable
-autoincrement*. This is divergent with the majority of documentation
+older Oracle versions, *a sequence must always be explicitly specified to
+enable autoincrement*. This is divergent with the majority of documentation
examples which assume the usage of an autoincrement-capable database. To
specify sequences, use the sqlalchemy.schema.Sequence object which is passed
to a Column construct::
autoload=True
)
+.. versionchanged:: 1.4 Added :class:`_schema.Identity` construct
+ in a :class:`_schema.Column` to specify the option of an autoincrementing
+ column.
+
Transaction Isolation Level / Autocommit
----------------------------------------
text += " VIRTUAL"
return text
+ def visit_identity_column(self, identity, **kw):
+ if identity.always is None:
+ kind = ""
+ else:
+ kind = "ALWAYS" if identity.always else "BY DEFAULT"
+ text = "GENERATED %s" % kind
+ if identity.on_null:
+ text += " ON NULL"
+ text += " AS IDENTITY"
+ options = self.get_identity_options(identity)
+ if options:
+ text += " (%s)" % options
+ return text
+
class OracleIdentifierPreparer(compiler.IdentifierPreparer):
To force the usage of RETURNING by default off, specify the flag
``implicit_returning=False`` to :func:`_sa.create_engine`.
-PostgreSQL 10 IDENTITY columns
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+PostgreSQL 10 and above IDENTITY columns
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-PostgreSQL 10 has a new IDENTITY feature that supersedes the use of SERIAL.
-Built-in support for rendering of IDENTITY is not available yet, however the
-following compilation hook may be used to replace occurrences of SERIAL with
-IDENTITY::
+PostgreSQL 10 and above have a new IDENTITY feature that supersedes the use
+of SERIAL. The :class:`_schema.Identity` construct in a
+:class:`_schema.Column` can be used to control its behavior::
- from sqlalchemy.schema import CreateColumn
- from sqlalchemy.ext.compiler import compiles
+ from sqlalchemy import Table, Column, MetaData, Integer, Computed
+ metadata = MetaData()
- @compiles(CreateColumn, 'postgresql')
- def use_identity(element, compiler, **kw):
- text = compiler.visit_create_column(element, **kw)
- text = text.replace("SERIAL", "INT GENERATED BY DEFAULT AS IDENTITY")
- return text
-
-Using the above, a table such as::
-
- t = Table(
- 't', m,
- Column('id', Integer, primary_key=True),
+ data = Table(
+ "data",
+ metadata,
+ Column(
+ 'id', Integer, Identity(start=42, cycle=True), primary_key=True
+ ),
Column('data', String)
)
-Will generate on the backing database as::
+The CREATE TABLE for the above :class:`_schema.Table` object would be:
+
+.. sourcecode:: sql
- CREATE TABLE t (
- id INT GENERATED BY DEFAULT AS IDENTITY NOT NULL,
+ CREATE TABLE data (
+ id INTEGER GENERATED BY DEFAULT AS IDENTITY (START WITH 42 CYCLE)
+ NOT NULL,
data VARCHAR,
PRIMARY KEY (id)
)
+.. versionchanged:: 1.4 Added :class:`_schema.Identity` construct
+ in a :class:`_schema.Column` to specify the option of an autoincrementing
+ column.
+
+.. note::
+
+ Previous versions of SQLAlchemy did not have built-in support for rendering
+ of IDENTITY, and could use the following compilation hook to replace
+ occurrences of SERIAL with IDENTITY::
+
+ from sqlalchemy.schema import CreateColumn
+ from sqlalchemy.ext.compiler import compiles
+
+
+ @compiles(CreateColumn, 'postgresql')
+ def use_identity(element, compiler, **kw):
+ text = compiler.visit_create_column(element, **kw)
+ text = text.replace(
+ "SERIAL", "INT GENERATED BY DEFAULT AS IDENTITY"
+ )
+ return text
+
+ Using the above, a table such as::
+
+ t = Table(
+ 't', m,
+ Column('id', Integer, primary_key=True),
+ Column('data', String)
+ )
+
+ Will generate on the backing database as::
+
+ CREATE TABLE t (
+ id INT GENERATED BY DEFAULT AS IDENTITY NOT NULL,
+ data VARCHAR,
+ PRIMARY KEY (id)
+ )
+
.. _postgresql_isolation_level:
Transaction Isolation Level
self.dialect.supports_smallserial
or not isinstance(impl_type, sqltypes.SmallInteger)
)
+ and column.identity is None
and (
column.default is None
or (
if column.computed is not None:
colspec += " " + self.process(column.computed)
+ if column.identity is not None:
+ colspec += " " + self.process(column.identity)
if not column.nullable:
colspec += " NOT NULL"
from .sql.schema import ForeignKey # noqa
from .sql.schema import ForeignKeyConstraint # noqa
from .sql.schema import Index # noqa
-from .sql.schema import IdentityOptions # noqa
+from .sql.schema import Identity # noqa
from .sql.schema import MetaData # noqa
from .sql.schema import PrimaryKeyConstraint # noqa
from .sql.schema import SchemaItem # noqa
drop.element, use_table=True
)
+ def get_identity_options(self, identity_options):
+ text = []
+ if identity_options.increment is not None:
+ text.append("INCREMENT BY %d" % identity_options.increment)
+ if identity_options.start is not None:
+ text.append("START WITH %d" % identity_options.start)
+ if identity_options.minvalue is not None:
+ text.append("MINVALUE %d" % identity_options.minvalue)
+ if identity_options.maxvalue is not None:
+ text.append("MAXVALUE %d" % identity_options.maxvalue)
+ if identity_options.nominvalue is not None:
+ text.append("NO MINVALUE")
+ if identity_options.nomaxvalue is not None:
+ text.append("NO MAXVALUE")
+ if identity_options.cache is not None:
+ text.append("CACHE %d" % identity_options.cache)
+ if identity_options.order is True:
+ text.append("ORDER")
+ if identity_options.cycle is not None:
+ text.append("CYCLE")
+ return " ".join(text)
+
def visit_create_sequence(self, create, prefix=None, **kw):
text = "CREATE SEQUENCE %s" % self.preparer.format_sequence(
create.element
)
if prefix:
text += prefix
- if create.element.increment is not None:
- text += " INCREMENT BY %d" % create.element.increment
if create.element.start is None:
create.element.start = self.dialect.default_sequence_base
- text += " START WITH %d" % create.element.start
- if create.element.minvalue is not None:
- text += " MINVALUE %d" % create.element.minvalue
- if create.element.maxvalue is not None:
- text += " MAXVALUE %d" % create.element.maxvalue
- if create.element.nominvalue is not None:
- text += " NO MINVALUE"
- if create.element.nomaxvalue is not None:
- text += " NO MAXVALUE"
- if create.element.cache is not None:
- text += " CACHE %d" % create.element.cache
- if create.element.order is True:
- text += " ORDER"
- if create.element.cycle is not None:
- text += " CYCLE"
+ options = self.get_identity_options(create.element)
+ if options:
+ text += " " + options
return text
def visit_drop_sequence(self, drop, **kw):
if column.computed is not None:
colspec += " " + self.process(column.computed)
+ if column.identity is not None:
+ colspec += " " + self.process(column.identity)
+
if not column.nullable:
colspec += " NOT NULL"
return colspec
text += " VIRTUAL"
return text
+ def visit_identity_column(self, identity, **kw):
+ text = "GENERATED %s AS IDENTITY" % (
+ "ALWAYS" if identity.always else "BY DEFAULT",
+ )
+ options = self.get_identity_options(identity)
+ if options:
+ text += " (%s)" % options
+ return text
+
class GenericTypeCompiler(TypeCompiler):
def visit_FLOAT(self, type_, **kw):
:class:`.SchemaItem` derived constructs which will be applied
as options to the column. These include instances of
:class:`.Constraint`, :class:`_schema.ForeignKey`,
- :class:`.ColumnDefault`,
- :class:`.Sequence`, :class:`.Computed`. In some cases an
+ :class:`.ColumnDefault`, :class:`.Sequence`, :class:`.Computed`
+ :class:`.Identity`. In some cases an
equivalent keyword argument is available such as ``server_default``,
``default`` and ``unique``.
AUTO_INCREMENT will be emitted for this column during a table
create, as well as that the column is assumed to generate new
integer primary key values when an INSERT statement invokes which
- will be retrieved by the dialect.
+ will be retrieved by the dialect. When used in conjunction with
+ :class:`.Identity` on a dialect that supports it, this parameter
+ has no effect.
The flag may be set to ``True`` to indicate that a column which
is part of a composite (e.g. multi-column) primary key should
self.foreign_keys = set()
self.comment = kwargs.pop("comment", None)
self.computed = None
+ self.identity = None
# check if this Column is proxying another column
if "_proxies" in kwargs:
self._setup_on_memoized_fks(lambda fk: fk._set_remote_table(table))
+ if self.identity and (
+ isinstance(self.default, Sequence)
+ or isinstance(self.onupdate, Sequence)
+ ):
+ raise exc.ArgumentError(
+ "An column cannot specify both Identity and Sequence."
+ )
+
def _setup_on_memoized_fks(self, fn):
fk_keys = [
((self.table.key, self.key), False),
server_default = self.server_default
server_onupdate = self.server_onupdate
- if isinstance(server_default, Computed):
+ if isinstance(server_default, (Computed, Identity)):
server_default = server_onupdate = None
args.append(self.server_default.copy(**kw))
"""Construct a :class:`.IdentityOptions` object.
See the :class:`.Sequence` documentation for a complete description
- of the parameters
+ of the parameters.
:param start: the starting index of the sequence.
:param increment: the increment value of the sequence.
and not autoinc_true
):
return False
- elif col.server_default is not None and not autoinc_true:
+ elif (
+ col.server_default is not None
+ and not isinstance(col.server_default, Identity)
+ and not autoinc_true
+ ):
return False
elif col.foreign_keys and col.autoincrement not in (
True,
g = Computed(sqltext, persisted=self.persisted)
return self._schema_item_copy(g)
+
+
+class Identity(IdentityOptions, FetchedValue, SchemaItem):
+ """Defines an identity column, i.e. "GENERATED { ALWAYS | BY DEFAULT }
+ AS IDENTITY" syntax.
+
+ The :class:`.Identity` construct is an inline construct added to the
+ argument list of a :class:`_schema.Column` object::
+
+ from sqlalchemy import Identity
+
+ Table('foo', meta,
+ Column('id', Integer, Identity())
+ Column('description', Text),
+ )
+
+ See the linked documentation below for complete details.
+
+ .. versionadded:: 1.4
+
+ .. seealso::
+
+ :ref:`identity_ddl`
+
+ """
+
+ __visit_name__ = "identity_column"
+
+ def __init__(
+ self,
+ always=False,
+ on_null=None,
+ start=None,
+ increment=None,
+ minvalue=None,
+ maxvalue=None,
+ nominvalue=None,
+ nomaxvalue=None,
+ cycle=None,
+ cache=None,
+ order=None,
+ ):
+ """Construct a GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY DDL
+ construct to accompany a :class:`_schema.Column`.
+
+ See the :class:`.Sequence` documentation for a complete description
+ of most parameters.
+
+ .. note::
+ MSSQL supports this construct as the preferred alternative to
+ generate an IDENTITY on a column, but it uses non standard
+ syntax that only support :paramref:`_schema.Identity.start`
+ and :paramref:`_schema.Identity.increment`.
+ All other parameters are ignored.
+
+ :param always:
+ A boolean, that indicates the type of identity column.
+ If ``False`` is specified, the default, then the user-specified
+ value takes precedence.
+ If ``True`` is specified, a user-specified value is not accepted (
+ on some backends, like PostgreSQL, OVERRIDING SYSTEM VALUE, or
+ similar, may be specified in an INSERT to override the sequence
+ value).
+ Some backends also have a default value for this parameter,
+ ``None`` can be used to omit rendering this part in the DDL. It
+ will be treated as ``False`` if a backend does not have a default
+ value.
+
+ :param on_null:
+ Set to ``True`` to specify ON NULL in conjunction with a
+ 'by default' identity column. This option is only supported on
+ some backends, like Oracle.
+
+ :param start: the starting index of the sequence.
+ :param increment: the increment value of the sequence.
+ :param minvalue: the minimum value of the sequence.
+ :param maxvalue: the maximum value of the sequence.
+ :param nominvalue: no minimum value of the sequence.
+ :param nomaxvalue: no maximum value of the sequence.
+ :param cycle: allows the sequence to wrap around when the maxvalue
+ or minvalue has been reached.
+ :param cache: optional integer value; number of future values in the
+ sequence which are calculated in advance.
+ :param order: optional boolean value; if true, renders the
+ ORDER keyword.
+
+ """
+ IdentityOptions.__init__(
+ self,
+ start=start,
+ increment=increment,
+ minvalue=minvalue,
+ maxvalue=maxvalue,
+ nominvalue=nominvalue,
+ nomaxvalue=nomaxvalue,
+ cycle=cycle,
+ cache=cache,
+ order=order,
+ )
+ self.always = always
+ self.on_null = on_null
+ self.column = None
+
+ def _set_parent(self, parent):
+ if not isinstance(
+ parent.server_default, (type(None), Identity)
+ ) or not isinstance(parent.server_onupdate, type(None)):
+ raise exc.ArgumentError(
+ "A column with an Identity object cannot specify a "
+ "server_default or a server_onupdate argument"
+ )
+ if parent.autoincrement is False:
+ raise exc.ArgumentError(
+ "A column with an Identity object cannot specify "
+ "autoincrement=False"
+ )
+ self.column = parent
+ parent.identity = self
+ # self.column.server_onupdate = self
+ self.column.server_default = self
+
+ def _as_for_update(self, for_update):
+ return self
+
+ def copy(self, target_table=None, **kw):
+ i = Identity(
+ always=self.always,
+ on_null=self.on_null,
+ start=self.start,
+ increment=self.increment,
+ minvalue=self.minvalue,
+ maxvalue=self.maxvalue,
+ nominvalue=self.nominvalue,
+ nomaxvalue=self.nomaxvalue,
+ cycle=self.cycle,
+ cache=self.cache,
+ order=self.order,
+ )
+
+ return self._schema_item_copy(i)
if inspect.isclass(obj) and plugin_base.want_class(name, obj):
- # in pytest 5.4.0
- # return [
- # pytest.Class.from_parent(collector,
- # name=parametrize_cls.__name__)
- # for parametrize_cls in _parametrize_cls(collector.module, obj)
- # ]
+ ctor = getattr(pytest.Class, "from_parent", pytest.Class)
return [
- pytest.Class(parametrize_cls.__name__, parent=collector)
+ ctor(name=parametrize_cls.__name__, parent=collector)
for parametrize_cls in _parametrize_cls(collector.module, obj)
]
elif (
lambda config: not config.db.dialect.supports_is_distinct_from,
"driver doesn't support an IS DISTINCT FROM construct",
)
+
+ @property
+ def identity_columns(self):
+ """If a backend supports GENERATED { ALWAYS | BY DEFAULT }
+ AS IDENTITY"""
+ return exclusions.closed()
+
+ @property
+ def identity_columns_standard(self):
+ """If a backend supports GENERATED { ALWAYS | BY DEFAULT }
+ AS IDENTITY with a standard syntax.
+ This is mainly to exclude MSSql.
+ """
+ return exclusions.closed()
from .. import AssertsExecutionResults
from .. import config
from .. import fixtures
+from ..assertions import assert_raises
from ..assertions import eq_
from ..assertions import in_
from ..assertsql import CursorSQL
from ... import false
from ... import ForeignKey
from ... import func
+from ... import Identity
from ... import Integer
from ... import literal
from ... import literal_column
from ... import tuple_
from ... import union
from ... import util
+from ...exc import DatabaseError
+from ...exc import ProgrammingError
class CollateTest(fixtures.TablesTest):
eq_(res, [(100, 40), (1764, 168)])
+class IdentityColumnTest(fixtures.TablesTest):
+ __backend__ = True
+ __requires__ = ("identity_columns",)
+ run_inserts = "once"
+ run_deletes = "once"
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "tbl_a",
+ metadata,
+ Column(
+ "id",
+ Integer,
+ Identity(always=True, start=42),
+ primary_key=True,
+ ),
+ Column("desc", String(100)),
+ )
+ Table(
+ "tbl_b",
+ metadata,
+ Column(
+ "id",
+ Integer,
+ Identity(increment=-5, start=0, minvalue=-1000, maxvalue=0,),
+ primary_key=True,
+ ),
+ Column("desc", String(100)),
+ )
+
+ @classmethod
+ def insert_data(cls, connection):
+ connection.execute(
+ cls.tables.tbl_a.insert(), [{"desc": "a"}, {"desc": "b"}],
+ )
+ connection.execute(
+ cls.tables.tbl_b.insert(), [{"desc": "a"}, {"desc": "b"}],
+ )
+ connection.execute(
+ cls.tables.tbl_b.insert(), [{"id": 42, "desc": "c"}],
+ )
+
+ def test_select_all(self, connection):
+ res = connection.execute(
+ select([text("*")])
+ .select_from(self.tables.tbl_a)
+ .order_by(self.tables.tbl_a.c.id)
+ ).fetchall()
+ eq_(res, [(42, "a"), (43, "b")])
+
+ res = connection.execute(
+ select([text("*")])
+ .select_from(self.tables.tbl_b)
+ .order_by(self.tables.tbl_b.c.id)
+ ).fetchall()
+ eq_(res, [(-5, "b"), (0, "a"), (42, "c")])
+
+ def test_select_columns(self, connection):
+
+ res = connection.execute(
+ select([self.tables.tbl_a.c.id]).order_by(self.tables.tbl_a.c.id)
+ ).fetchall()
+ eq_(res, [(42,), (43,)])
+
+ @testing.requires.identity_columns_standard
+ def test_insert_always_error(self, connection):
+ def fn():
+ connection.execute(
+ self.tables.tbl_a.insert(), [{"id": 200, "desc": "a"}],
+ )
+
+ assert_raises((DatabaseError, ProgrammingError), fn)
+
+
class ExistsTest(fixtures.TablesTest):
__backend__ = True
class DistinctOnTest(AssertsCompiledSQL, fixtures.TablesTest):
__backend__ = True
- __requires__ = ("standard_cursor_sql",)
@testing.fails_if(testing.requires.supports_distinct_on)
def test_distinct_on(self):
from sqlalchemy import Column
from sqlalchemy import Computed
from sqlalchemy import delete
+from sqlalchemy import exc
from sqlalchemy import extract
from sqlalchemy import func
+from sqlalchemy import Identity
from sqlalchemy import Index
from sqlalchemy import insert
from sqlalchemy import Integer
from sqlalchemy.sql import column
from sqlalchemy.sql import quoted_name
from sqlalchemy.sql import table
+from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
checkparams={"x_1": 5, "param_1": 0},
)
- def test_primary_key_no_identity(self):
- metadata = MetaData()
- tbl = Table(
- "test",
- metadata,
- Column("id", Integer, autoincrement=False, primary_key=True),
- )
- self.assert_compile(
- schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL, " "PRIMARY KEY (id))",
- )
-
- def test_primary_key_defaults_to_identity(self):
- metadata = MetaData()
- tbl = Table("test", metadata, Column("id", Integer, primary_key=True))
- self.assert_compile(
- schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1), "
- "PRIMARY KEY (id))",
- )
-
- def test_identity_no_primary_key(self):
- metadata = MetaData()
- tbl = Table(
- "test", metadata, Column("id", Integer, autoincrement=True)
- )
- self.assert_compile(
- schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1)" ")",
- )
-
- def test_identity_separate_from_primary_key(self):
- metadata = MetaData()
- tbl = Table(
- "test",
- metadata,
- Column("id", Integer, autoincrement=False, primary_key=True),
- Column("x", Integer, autoincrement=True),
- )
- self.assert_compile(
- schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL, "
- "x INTEGER NOT NULL IDENTITY(1,1), "
- "PRIMARY KEY (id))",
- )
-
- def test_identity_illegal_two_autoincrements(self):
- metadata = MetaData()
- tbl = Table(
- "test",
- metadata,
- Column("id", Integer, autoincrement=True),
- Column("id2", Integer, autoincrement=True),
- )
- # this will be rejected by the database, just asserting this is what
- # the two autoincrements will do right now
- self.assert_compile(
- schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1), "
- "id2 INTEGER NOT NULL IDENTITY(1,1))",
- )
-
- def test_identity_start_0(self):
- metadata = MetaData()
- tbl = Table(
- "test",
- metadata,
- Column("id", Integer, mssql_identity_start=0, primary_key=True),
- )
- self.assert_compile(
- schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), "
- "PRIMARY KEY (id))",
- )
-
- def test_identity_increment_5(self):
- metadata = MetaData()
- tbl = Table(
- "test",
- metadata,
- Column(
- "id", Integer, mssql_identity_increment=5, primary_key=True
- ),
- )
- self.assert_compile(
- schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,5), "
- "PRIMARY KEY (id))",
- )
-
def test_table_pkc_clustering(self):
metadata = MetaData()
tbl = Table(
)
+class CompileIdentityTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = mssql.dialect()
+
+ def assert_compile_with_warning(self, *args, **kwargs):
+ with testing.expect_deprecated(
+ "The dialect options 'mssql_identity_start' and "
+ "'mssql_identity_increment' are deprecated. "
+ "Use the 'Identity' object instead."
+ ):
+ return self.assert_compile(*args, **kwargs)
+
+ def test_primary_key_no_identity(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column("id", Integer, autoincrement=False, primary_key=True),
+ )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL, PRIMARY KEY (id))",
+ )
+
+ def test_primary_key_defaults_to_identity(self):
+ metadata = MetaData()
+ tbl = Table("test", metadata, Column("id", Integer, primary_key=True))
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY, "
+ "PRIMARY KEY (id))",
+ )
+
+ def test_primary_key_with_identity_object(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column(
+ "id",
+ Integer,
+ Identity(start=3, increment=42),
+ primary_key=True,
+ ),
+ )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,42), "
+ "PRIMARY KEY (id))",
+ )
+
+ def test_identity_no_primary_key(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test", metadata, Column("id", Integer, autoincrement=True)
+ )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY)",
+ )
+
+ def test_identity_object_no_primary_key(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test", metadata, Column("id", Integer, Identity(increment=42)),
+ )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,42))",
+ )
+
+ def test_identity_object_1_1(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column("id", Integer, Identity(start=1, increment=1)),
+ )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1))",
+ )
+
+ def test_identity_object_no_primary_key_non_nullable(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column("id", Integer, Identity(start=3), nullable=False,),
+ )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,1)" ")",
+ )
+
+ def test_identity_separate_from_primary_key(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column("id", Integer, autoincrement=False, primary_key=True),
+ Column("x", Integer, autoincrement=True),
+ )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL, "
+ "x INTEGER NOT NULL IDENTITY, "
+ "PRIMARY KEY (id))",
+ )
+
+ def test_identity_object_separate_from_primary_key(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column("id", Integer, autoincrement=False, primary_key=True),
+ Column("x", Integer, Identity(start=3, increment=42),),
+ )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL, "
+ "x INTEGER NOT NULL IDENTITY(3,42), "
+ "PRIMARY KEY (id))",
+ )
+
+ def test_identity_illegal_two_autoincrements(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column("id", Integer, autoincrement=True),
+ Column("id2", Integer, autoincrement=True),
+ )
+ # this will be rejected by the database, just asserting this is what
+ # the two autoincrements will do right now
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY, "
+ "id2 INTEGER NOT NULL IDENTITY)",
+ )
+
+ def test_identity_object_illegal_two_autoincrements(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column(
+ "id",
+ Integer,
+ Identity(start=3, increment=42),
+ autoincrement=True,
+ ),
+ Column("id2", Integer, Identity(start=7, increment=2),),
+ )
+ # this will be rejected by the database, just asserting this is what
+ # the two autoincrements will do right now
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,42), "
+ "id2 INTEGER NOT NULL IDENTITY(7,2))",
+ )
+
+ def test_identity_start_0(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column("id", Integer, mssql_identity_start=0, primary_key=True),
+ )
+ self.assert_compile_with_warning(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), "
+ "PRIMARY KEY (id))",
+ )
+
+ def test_identity_increment_5(self):
+ metadata = MetaData()
+ tbl = Table(
+ "test",
+ metadata,
+ Column(
+ "id", Integer, mssql_identity_increment=5, primary_key=True
+ ),
+ )
+ self.assert_compile_with_warning(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,5), "
+ "PRIMARY KEY (id))",
+ )
+
+ @testing.combinations(
+ schema.CreateTable(
+ Table(
+ "test",
+ MetaData(),
+ Column(
+ "id",
+ Integer,
+ Identity(start=2, increment=2),
+ mssql_identity_start=0,
+ ),
+ )
+ ),
+ schema.CreateTable(
+ Table(
+ "test1",
+ MetaData(),
+ Column(
+ "id2",
+ Integer,
+ Identity(start=3, increment=3),
+ mssql_identity_increment=5,
+ ),
+ )
+ ),
+ )
+ def test_identity_options_ignored_with_identity_object(self, create_table):
+ assert_raises_message(
+ exc.CompileError,
+ "Cannot specify options 'mssql_identity_start' and/or "
+ "'mssql_identity_increment' while also using the "
+ "'Identity' construct.",
+ create_table.compile,
+ dialect=self.__dialect__,
+ )
+
+ def test_identity_object_no_options(self):
+ metadata = MetaData()
+ tbl = Table("test", metadata, Column("id", Integer, Identity()),)
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY)",
+ )
+
+
class SchemaTest(fixtures.TestBase):
def setup(self):
t = Table(
from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import func
+from sqlalchemy import Identity
from sqlalchemy import Integer
from sqlalchemy import literal
from sqlalchemy import MetaData
t1 = Table(
"t1",
meta,
- Column("id", Integer, mssql_identity_start=100, primary_key=True),
+ Column("id", Integer, Identity(start=100), primary_key=True),
Column("descr", String(200)),
# the following flag will prevent the
# MSSQLCompiler.returning_clause from getting called,
t2 = Table(
"t2",
meta,
- Column("id", Integer, mssql_identity_start=200, primary_key=True),
+ Column("id", Integer, Identity(start=200), primary_key=True),
Column("descr", String(200)),
)
primary_key=True,
),
)
- table.create()
+ with testing.expect_deprecated(
+ "The dialect options 'mssql_identity_start' and"
+ ):
+ table.create()
meta2 = MetaData(testing.db)
table2 = Table("identity_test", meta2, autoload=True)
from sqlalchemy import except_
from sqlalchemy import ForeignKey
from sqlalchemy import func
+from sqlalchemy import Identity
from sqlalchemy import Index
from sqlalchemy import Integer
from sqlalchemy import literal
dialect=oracle.dialect(),
)
+ def test_column_identity(self):
+ # all other tests are in test_identity_column.py
+ m = MetaData()
+ t = Table(
+ "t",
+ m,
+ Column("y", Integer, Identity(always=True, start=4, increment=7)),
+ )
+ self.assert_compile(
+ schema.CreateTable(t),
+ "CREATE TABLE t (y INTEGER GENERATED ALWAYS AS IDENTITY "
+ "(INCREMENT BY 7 START WITH 4))",
+ )
+
+ def test_column_identity_no_generated(self):
+ m = MetaData()
+ t = Table("t", m, Column("y", Integer, Identity(always=None)))
+ self.assert_compile(
+ schema.CreateTable(t),
+ "CREATE TABLE t (y INTEGER GENERATED AS IDENTITY)",
+ )
+
+ @testing.combinations(
+ (True, True, "ALWAYS ON NULL"), # this would error when executed
+ (False, None, "BY DEFAULT"),
+ (False, False, "BY DEFAULT"),
+ (False, True, "BY DEFAULT ON NULL"),
+ )
+ def test_column_identity_on_null(self, always, on_null, text):
+ m = MetaData()
+ t = Table(
+ "t", m, Column("y", Integer, Identity(always, on_null=on_null))
+ )
+ self.assert_compile(
+ schema.CreateTable(t),
+ "CREATE TABLE t (y INTEGER GENERATED %s AS IDENTITY)" % text,
+ )
+
class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):
def test_basic(self):
from sqlalchemy import Enum
from sqlalchemy import exc
from sqlalchemy import func
+from sqlalchemy import Identity
from sqlalchemy import Index
from sqlalchemy import Integer
from sqlalchemy import MetaData
dialect=postgresql.dialect(),
)
+ def test_column_identity(self):
+ # all other tests are in test_identity_column.py
+ m = MetaData()
+ t = Table(
+ "t",
+ m,
+ Column("y", Integer, Identity(always=True, start=4, increment=7)),
+ )
+ self.assert_compile(
+ schema.CreateTable(t),
+ "CREATE TABLE t (y INTEGER GENERATED ALWAYS AS IDENTITY "
+ "(INCREMENT BY 7 START WITH 4))",
+ )
+
class InsertOnConflictTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = postgresql.dialect()
import sqlalchemy as sa
from sqlalchemy import Computed
from sqlalchemy import event
+from sqlalchemy import Identity
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import testing
[{"param_1": 2}],
),
)
+
+
+class IdentityDefaultsOnUpdateTest(fixtures.MappedTest):
+ """test that computed columns are recognized as server
+ oninsert/onupdate defaults."""
+
+ __backend__ = True
+ __requires__ = ("identity_columns",)
+ run_create_tables = "each"
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "test",
+ metadata,
+ Column("id", Integer, Identity(), primary_key=True),
+ Column("foo", Integer),
+ )
+
+ @classmethod
+ def setup_classes(cls):
+ class Thing(cls.Basic):
+ pass
+
+ @classmethod
+ def setup_mappers(cls):
+ Thing = cls.classes.Thing
+
+ mapper(Thing, cls.tables.test)
+
+ def test_insert_identity(self):
+ Thing = self.classes.Thing
+
+ s = Session()
+
+ t1, t2 = (Thing(foo=5), Thing(foo=10))
+
+ s.add_all([t1, t2])
+
+ with assert_engine(testing.db) as asserter:
+ s.flush()
+ eq_(t1.id, 1)
+ eq_(t2.id, 2)
+
+ asserter.assert_(
+ Conditional(
+ testing.db.dialect.implicit_returning,
+ [
+ Conditional(
+ testing.db.dialect.insert_executemany_returning,
+ [
+ CompiledSQL(
+ "INSERT INTO test (foo) VALUES (%(foo)s) "
+ "RETURNING test.id",
+ [{"foo": 5}, {"foo": 10}],
+ dialect="postgresql",
+ ),
+ ],
+ [
+ CompiledSQL(
+ "INSERT INTO test (foo) VALUES (%(foo)s) "
+ "RETURNING test.id",
+ [{"foo": 5}],
+ dialect="postgresql",
+ ),
+ CompiledSQL(
+ "INSERT INTO test (foo) VALUES (%(foo)s) "
+ "RETURNING test.id",
+ [{"foo": 10}],
+ dialect="postgresql",
+ ),
+ ],
+ )
+ ],
+ [
+ CompiledSQL(
+ "INSERT INTO test (foo) VALUES (:foo)", [{"foo": 5}],
+ ),
+ CompiledSQL(
+ "INSERT INTO test (foo) VALUES (:foo)", [{"foo": 10}],
+ ),
+ ],
+ )
+ )
def supports_sequence_for_autoincrement_column(self):
"""for mssql, autoincrement means IDENTITY, not sequence"""
return skip_if("mssql")
+
+ @property
+ def identity_columns(self):
+ return only_if(["postgresql >= 10", "oracle >= 12", "mssql"])
+
+ @property
+ def identity_columns_standard(self):
+ return self.identity_columns + skip_if("mssql")
"ALWAYS AS (x + 2)%s)" % text,
)
+ def test_other_options(self):
+ t = Table(
+ "t",
+ MetaData(),
+ Column(
+ "y", Integer, Computed("x + 2"), nullable=False, unique=True
+ ),
+ )
+ self.assert_compile(
+ CreateTable(t),
+ "CREATE TABLE t ("
+ "y INTEGER GENERATED ALWAYS AS (x + 2) NOT NULL, UNIQUE (y))",
+ )
+
def test_server_default_onupdate(self):
text = (
"A generated column cannot specify a server_default or a "
--- /dev/null
+import re
+
+from sqlalchemy import Column
+from sqlalchemy import Identity
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import Sequence
+from sqlalchemy import Table
+from sqlalchemy import testing
+from sqlalchemy.exc import ArgumentError
+from sqlalchemy.schema import CreateTable
+from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import is_
+from sqlalchemy.testing import is_not_
+
+
+class _IdentityDDLFixture(testing.AssertsCompiledSQL):
+ __backend__ = True
+
+ @testing.combinations(
+ (dict(always=True), "ALWAYS AS IDENTITY"),
+ (
+ dict(always=False, start=5),
+ "BY DEFAULT AS IDENTITY (START WITH 5)",
+ ),
+ (
+ dict(always=True, increment=2),
+ "ALWAYS AS IDENTITY (INCREMENT BY 2)",
+ ),
+ (
+ dict(increment=2, start=5),
+ "BY DEFAULT AS IDENTITY (INCREMENT BY 2 START WITH 5)",
+ ),
+ (
+ dict(always=True, increment=2, start=0, minvalue=0),
+ "ALWAYS AS IDENTITY (INCREMENT BY 2 START WITH 0 MINVALUE 0)",
+ ),
+ (
+ dict(always=False, increment=2, start=1, maxvalue=5),
+ "BY DEFAULT AS IDENTITY (INCREMENT BY 2 START WITH 1 MAXVALUE 5)",
+ ),
+ (
+ dict(always=True, increment=2, start=1, nomaxvalue=True),
+ "ALWAYS AS IDENTITY (INCREMENT BY 2 START WITH 1 NO MAXVALUE)",
+ ),
+ (
+ dict(always=False, increment=2, start=0, nominvalue=True),
+ "BY DEFAULT AS IDENTITY "
+ "(INCREMENT BY 2 START WITH 0 NO MINVALUE)",
+ ),
+ (
+ dict(always=True, start=1, maxvalue=10, cycle=True),
+ "ALWAYS AS IDENTITY (START WITH 1 MAXVALUE 10 CYCLE)",
+ ),
+ (
+ dict(always=False, cache=1000, order=True),
+ "BY DEFAULT AS IDENTITY (CACHE 1000 ORDER)",
+ ),
+ (dict(order=True), "BY DEFAULT AS IDENTITY (ORDER)",),
+ )
+ def test_create_ddl(self, identity_args, text):
+
+ t = Table(
+ "foo_table",
+ MetaData(),
+ Column("foo", Integer(), Identity(**identity_args)),
+ )
+ self.assert_compile(
+ CreateTable(t),
+ "CREATE TABLE foo_table (foo INTEGER GENERATED %s)" % text,
+ )
+ t2 = t.to_metadata(MetaData())
+ self.assert_compile(
+ CreateTable(t2),
+ "CREATE TABLE foo_table (foo INTEGER GENERATED %s)" % text,
+ )
+
+ def test_other_options(self):
+ t = Table(
+ "foo_table",
+ MetaData(),
+ Column(
+ "foo",
+ Integer(),
+ Identity(always=True, start=3),
+ nullable=False,
+ unique=True,
+ ),
+ )
+ self.assert_compile(
+ CreateTable(t),
+ "CREATE TABLE foo_table ("
+ "foo INTEGER GENERATED ALWAYS AS IDENTITY (START "
+ "WITH 3) NOT NULL, UNIQUE (foo))",
+ )
+
+ def test_autoincrement_true(self):
+ t = Table(
+ "foo_table",
+ MetaData(),
+ Column(
+ "foo",
+ Integer(),
+ Identity(always=True, start=3),
+ primary_key=True,
+ autoincrement=True,
+ ),
+ )
+ self.assert_compile(
+ CreateTable(t),
+ "CREATE TABLE foo_table ("
+ "foo INTEGER GENERATED ALWAYS AS IDENTITY (START WITH 3) NOT NULL"
+ ", PRIMARY KEY (foo))",
+ )
+
+
+class IdentityDDL(_IdentityDDLFixture, fixtures.TestBase):
+ # this uses the connection dialect
+ __requires__ = ("identity_columns_standard",)
+
+ def test_on_null(self):
+ t = Table(
+ "foo_table",
+ MetaData(),
+ Column(
+ "foo",
+ Integer(),
+ Identity(always=False, on_null=True, start=42, order=True),
+ ),
+ )
+ text = " ON NULL" if testing.against("oracle") else ""
+ self.assert_compile(
+ CreateTable(t),
+ (
+ "CREATE TABLE foo_table (foo INTEGER GENERATED BY DEFAULT"
+ + text
+ + " AS IDENTITY (START WITH 42 ORDER))"
+ ),
+ )
+
+
+class DefaultDialectIdentityDDL(_IdentityDDLFixture, fixtures.TestBase):
+ # this uses the default dialect
+ __dialect__ = "default"
+
+
+class NotSupportingIdentityDDL(testing.AssertsCompiledSQL, fixtures.TestBase):
+ @testing.skip_if(testing.requires.identity_columns)
+ def test_identity_is_ignored(self):
+ t = Table(
+ "foo_table",
+ MetaData(),
+ Column("foo", Integer(), Identity("always", start=3)),
+ )
+ t2 = Table("foo_table", MetaData(), Column("foo", Integer()),)
+ exp = CreateTable(t2).compile(dialect=testing.db.dialect)
+ self.assert_compile(
+ CreateTable(t), re.sub(r"[\n\t]", "", str(exp)),
+ )
+
+
+class IdentityTest(fixtures.TestBase):
+ def test_server_default_onupdate(self):
+ text = (
+ "A column with an Identity object cannot specify a "
+ "server_default or a server_onupdate argument"
+ )
+
+ def fn(**kwargs):
+ Table(
+ "t", MetaData(), Column("y", Integer, Identity(), **kwargs),
+ )
+
+ assert_raises_message(ArgumentError, text, fn, server_default="42")
+ assert_raises_message(ArgumentError, text, fn, server_onupdate="42")
+
+ def test_to_metadata(self):
+ identity1 = Identity("by default", on_null=True, start=123)
+ m = MetaData()
+ t = Table(
+ "t", m, Column("x", Integer), Column("y", Integer, identity1)
+ )
+ is_(identity1.column, t.c.y)
+ # is_(t.c.y.server_onupdate, identity1)
+ is_(t.c.y.server_default, identity1)
+
+ m2 = MetaData()
+ t2 = t.to_metadata(m2)
+ identity2 = t2.c.y.server_default
+
+ is_not_(identity1, identity2)
+
+ is_(identity1.column, t.c.y)
+ # is_(t.c.y.server_onupdate, identity1)
+ is_(t.c.y.server_default, identity1)
+
+ is_(identity2.column, t2.c.y)
+ # is_(t2.c.y.server_onupdate, identity2)
+ is_(t2.c.y.server_default, identity2)
+
+ def test_autoincrement_column(self):
+ t = Table(
+ "t",
+ MetaData(),
+ Column("y", Integer, Identity(), primary_key=True),
+ )
+ assert t._autoincrement_column is t.c.y
+ t2 = Table("t2", MetaData(), Column("y", Integer, Identity()))
+ assert t2._autoincrement_column is None
+
+ def test_identity_and_sequence(self):
+ def go():
+ return Table(
+ "foo_table",
+ MetaData(),
+ Column("foo", Integer(), Identity(), Sequence("foo_seq")),
+ )
+
+ assert_raises_message(
+ ArgumentError,
+ "An column cannot specify both Identity and Sequence.",
+ go,
+ )
+
+ def test_identity_autoincrement_false(self):
+ def go():
+ return Table(
+ "foo_table",
+ MetaData(),
+ Column("foo", Integer(), Identity(), autoincrement=False),
+ )
+
+ assert_raises_message(
+ ArgumentError,
+ "A column with an Identity object cannot specify "
+ "autoincrement=False",
+ go,
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
