Merge "Add support for identity columns"

2 files changed, 253 insertions, 0 deletions

diff --git a/test/sql/test_computed.py b/test/sql/test_computed.py
index c7cb3dd11..56aac488f 100644
--- a/test/sql/test_computed.py
+++ b/test/sql/test_computed.py
@@ -39,6 +39,20 @@ class DDLComputedTest(fixtures.TestBase, AssertsCompiledSQL):
             "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 "
diff --git a/test/sql/test_identity_column.py b/test/sql/test_identity_column.py
new file mode 100644
index 000000000..becb62159
--- /dev/null
+++ b/test/sql/test_identity_column.py
@@ -0,0 +1,239 @@
+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,
+        )