Add support identity columns.

Added support for rendering of "identity" elements on
:class:`.Column` objects, supported in SQLAlchemy via
the :class:`.Identity` element introduced in version 1.4.

Adding columns with identity is supported on PostgreSQL,
MSSQL and Oracle. Changing the identity options or removing
it is supported only on PostgreSQL and Oracle.

Fixes: #730
Change-Id: I184d8bd30ef5c6286f3263b4033ca4eb359c02e2

8 files changed, 780 insertions, 14 deletions

diff --git a/tests/requirements.py b/tests/requirements.py
index eb6066d..6f171f6 100644
--- a/tests/requirements.py
+++ b/tests/requirements.py
@@ -251,3 +251,23 @@ class DefaultRequirements(SuiteRequirements):
                 return norm_version_info >= (8, 0, 16)
         else:
             return False
+
+    @property
+    def identity_columns(self):
+        # TODO: in theory if these could come from SQLAlchemy dialects
+        # that would be helpful
+        return self.identity_columns_api + exclusions.only_on(
+            ["postgresql >= 10", "oracle >= 12", "mssql"]
+        )
+
+    @property
+    def identity_columns_alter(self):
+        # TODO: in theory if these could come from SQLAlchemy dialects
+        # that would be helpful
+        return self.identity_columns_api + exclusions.only_on(
+            ["postgresql >= 10", "oracle >= 12"]
+        )
+
+    @property
+    def supports_identity_on_null(self):
+        return self.identity_columns + exclusions.only_on(["oracle"])
diff --git a/tests/test_autogen_identity.py b/tests/test_autogen_identity.py
new file mode 100644
index 0000000..9d8253a
--- /dev/null
+++ b/tests/test_autogen_identity.py
@@ -0,0 +1,241 @@
+import sqlalchemy as sa
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import Table
+
+from alembic import testing
+from alembic.testing import config
+from alembic.testing import eq_
+from alembic.testing import is_true
+from alembic.testing import TestBase
+from ._autogen_fixtures import AutogenFixtureTest
+
+
+class AutogenerateIdentityTest(AutogenFixtureTest, TestBase):
+    __requires__ = ("identity_columns",)
+    __backend__ = True
+
+    def test_add_identity_column(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table("user", m1, Column("other", sa.Text))
+
+        Table(
+            "user",
+            m2,
+            Column("other", sa.Text),
+            Column(
+                "id",
+                Integer,
+                sa.Identity(start=5, increment=7),
+                primary_key=True,
+            ),
+        )
+
+        diffs = self._fixture(m1, m2)
+
+        eq_(diffs[0][0], "add_column")
+        eq_(diffs[0][2], "user")
+        eq_(diffs[0][3].name, "id")
+        i = diffs[0][3].identity
+
+        is_true(isinstance(i, sa.Identity))
+        eq_(i.start, 5)
+        eq_(i.increment, 7)
+
+    def test_remove_identity_column(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table(
+            "user",
+            m1,
+            Column(
+                "id",
+                Integer,
+                sa.Identity(start=2, increment=3),
+                primary_key=True,
+            ),
+        )
+
+        Table("user", m2)
+
+        diffs = self._fixture(m1, m2)
+
+        eq_(diffs[0][0], "remove_column")
+        eq_(diffs[0][2], "user")
+        c = diffs[0][3]
+        eq_(c.name, "id")
+
+        is_true(isinstance(c.identity, sa.Identity))
+        eq_(c.identity.start, 2)
+        eq_(c.identity.increment, 3)
+
+    def test_no_change_identity_column(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        for m in (m1, m2):
+            Table(
+                "user",
+                m,
+                Column("id", Integer, sa.Identity(start=2)),
+            )
+
+        diffs = self._fixture(m1, m2)
+
+        eq_(diffs, [])
+
+    @testing.combinations(
+        (None, dict(start=2)),
+        (dict(start=2), None),
+        (dict(start=2), dict(start=2, increment=7)),
+        (dict(always=False), dict(always=True)),
+        (
+            dict(start=1, minvalue=0, maxvalue=100, cycle=True),
+            dict(start=1, minvalue=0, maxvalue=100, cycle=False),
+        ),
+        (
+            dict(start=10, increment=3, maxvalue=9999),
+            dict(start=10, increment=1, maxvalue=3333),
+        ),
+    )
+    @config.requirements.identity_columns_alter
+    def test_change_identity(self, before, after):
+        arg_before = (sa.Identity(**before),) if before else ()
+        arg_after = (sa.Identity(**after),) if after else ()
+
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table(
+            "user",
+            m1,
+            Column("id", Integer, *arg_before),
+            Column("other", sa.Text),
+        )
+
+        Table(
+            "user",
+            m2,
+            Column("id", Integer, *arg_after),
+            Column("other", sa.Text),
+        )
+
+        diffs = self._fixture(m1, m2)
+
+        eq_(len(diffs[0]), 1)
+        diffs = diffs[0][0]
+        eq_(diffs[0], "modify_default")
+        eq_(diffs[2], "user")
+        eq_(diffs[3], "id")
+        old = diffs[5]
+        new = diffs[6]
+
+        def check(kw, idt):
+            if kw:
+                is_true(isinstance(idt, sa.Identity))
+                for k, v in kw.items():
+                    eq_(getattr(idt, k), v)
+            else:
+                is_true(idt in (None, False))
+
+        check(before, old)
+        check(after, new)
+
+    def test_add_identity_to_column(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table(
+            "user",
+            m1,
+            Column("id", Integer),
+            Column("other", sa.Text),
+        )
+
+        Table(
+            "user",
+            m2,
+            Column("id", Integer, sa.Identity(start=2, maxvalue=1000)),
+            Column("other", sa.Text),
+        )
+
+        diffs = self._fixture(m1, m2)
+
+        eq_(len(diffs[0]), 1)
+        diffs = diffs[0][0]
+        eq_(diffs[0], "modify_default")
+        eq_(diffs[2], "user")
+        eq_(diffs[3], "id")
+        eq_(diffs[5], None)
+        added = diffs[6]
+
+        is_true(isinstance(added, sa.Identity))
+        eq_(added.start, 2)
+        eq_(added.maxvalue, 1000)
+
+    def test_remove_identity_from_column(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table(
+            "user",
+            m1,
+            Column("id", Integer, sa.Identity(start=2, maxvalue=1000)),
+            Column("other", sa.Text),
+        )
+
+        Table(
+            "user",
+            m2,
+            Column("id", Integer),
+            Column("other", sa.Text),
+        )
+
+        diffs = self._fixture(m1, m2)
+
+        eq_(len(diffs[0]), 1)
+        diffs = diffs[0][0]
+        eq_(diffs[0], "modify_default")
+        eq_(diffs[2], "user")
+        eq_(diffs[3], "id")
+        eq_(diffs[6], None)
+        removed = diffs[5]
+
+        is_true(isinstance(removed, sa.Identity))
+
+    def test_identity_on_null(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table(
+            "user",
+            m1,
+            Column("id", Integer, sa.Identity(start=2, on_null=True)),
+            Column("other", sa.Text),
+        )
+
+        Table(
+            "user",
+            m2,
+            Column("id", Integer, sa.Identity(start=2, on_null=False)),
+            Column("other", sa.Text),
+        )
+
+        diffs = self._fixture(m1, m2)
+        if not config.requirements.supports_identity_on_null.enabled:
+            eq_(diffs, [])
+        else:
+            eq_(len(diffs[0]), 1)
+            diffs = diffs[0][0]
+            eq_(diffs[0], "modify_default")
+            eq_(diffs[2], "user")
+            eq_(diffs[3], "id")
+            old = diffs[5]
+            new = diffs[6]
+
+            is_true(isinstance(old, sa.Identity))
+            is_true(isinstance(new, sa.Identity))
diff --git a/tests/test_autogen_render.py b/tests/test_autogen_render.py
index e1f3269..d2dcc34 100644
--- a/tests/test_autogen_render.py
+++ b/tests/test_autogen_render.py
@@ -2123,6 +2123,93 @@ class AutogenRenderTest(TestBase):
             % persisted,
         )
 
+    @config.requirements.identity_columns
+    @testing.combinations(
+        ({}, "sa.Identity(always=False)"),
+        (dict(always=None), "sa.Identity(always=None)"),
+        (dict(always=True), "sa.Identity(always=True)"),
+        (
+            dict(
+                always=False,
+                on_null=True,
+                start=2,
+                increment=4,
+                minvalue=-3,
+                maxvalue=99,
+                nominvalue=True,
+                nomaxvalue=True,
+                cycle=True,
+                cache=42,
+                order=True,
+            ),
+            "sa.Identity(always=False, on_null=True, start=2, increment=4, "
+            "minvalue=-3, maxvalue=99, nominvalue=True, nomaxvalue=True, "
+            "cycle=True, cache=42, order=True)",
+        ),
+    )
+    def test_render_add_column_identity(self, kw, text):
+        op_obj = ops.AddColumnOp(
+            "foo", Column("x", Integer, sa.Identity(**kw))
+        )
+        eq_ignore_whitespace(
+            autogenerate.render_op_text(self.autogen_context, op_obj),
+            "op.add_column('foo', sa.Column('x', sa.Integer(), "
+            "%s, nullable=True))" % text,
+        )
+
+    @config.requirements.identity_columns
+    @testing.combinations(
+        ({}, "sa.Identity(always=False)"),
+        (dict(always=None), "sa.Identity(always=None)"),
+        (dict(always=True), "sa.Identity(always=True)"),
+        (
+            dict(
+                always=False,
+                on_null=True,
+                start=2,
+                increment=4,
+                minvalue=-3,
+                maxvalue=99,
+                nominvalue=True,
+                nomaxvalue=True,
+                cycle=True,
+                cache=42,
+                order=True,
+            ),
+            "sa.Identity(always=False, on_null=True, start=2, increment=4, "
+            "minvalue=-3, maxvalue=99, nominvalue=True, nomaxvalue=True, "
+            "cycle=True, cache=42, order=True)",
+        ),
+    )
+    def test_render_alter_column_add_identity(self, kw, text):
+        op_obj = ops.AlterColumnOp(
+            "foo",
+            "x",
+            existing_type=Integer(),
+            existing_server_default=None,
+            modify_server_default=sa.Identity(**kw),
+        )
+        eq_ignore_whitespace(
+            autogenerate.render_op_text(self.autogen_context, op_obj),
+            "op.alter_column('foo', 'x', existing_type=sa.Integer(), "
+            "server_default=%s)" % text,
+        )
+
+    @config.requirements.identity_columns
+    def test_render_alter_column_drop_identity(self):
+        op_obj = ops.AlterColumnOp(
+            "foo",
+            "x",
+            existing_type=Integer(),
+            existing_server_default=sa.Identity(),
+            modify_server_default=None,
+        )
+        eq_ignore_whitespace(
+            autogenerate.render_op_text(self.autogen_context, op_obj),
+            "op.alter_column('foo', 'x', existing_type=sa.Integer(), "
+            "server_default=None)",
+        )
+
 
 class RenderNamingConventionTest(TestBase):
     def setUp(self):
diff --git a/tests/test_mssql.py b/tests/test_mssql.py
index 36b90cb..0c8b3ae 100644
--- a/tests/test_mssql.py
+++ b/tests/test_mssql.py
@@ -1,12 +1,14 @@
 """Test op functions against MSSQL."""
 
 from sqlalchemy import Column
+from sqlalchemy import exc
 from sqlalchemy import Integer
 
 from alembic import command
 from alembic import op
 from alembic import util
 from alembic.testing import assert_raises_message
+from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing import eq_
 from alembic.testing.env import _no_sql_testing_config
@@ -354,3 +356,74 @@ class OpTest(TestBase):
         context.assert_contains(
             "CREATE INDEX ix_mytable_a_b ON mytable " "(col_a, col_b)"
         )
+
+    @combinations(
+        (lambda: sqla_compat.Computed("foo * 5"), lambda: None),
+        (lambda: None, lambda: sqla_compat.Computed("foo * 5")),
+        (
+            lambda: sqla_compat.Computed("foo * 42"),
+            lambda: sqla_compat.Computed("foo * 5"),
+        ),
+    )
+    @config.requirements.computed_columns
+    def test_alter_column_computed_not_supported(self, sd, esd):
+        op_fixture("mssql")
+        assert_raises_message(
+            exc.CompileError,
+            'Adding or removing a "computed" construct, e.g. '
+            "GENERATED ALWAYS AS, to or from an existing column is not "
+            "supported.",
+            op.alter_column,
+            "t1",
+            "c1",
+            server_default=sd(),
+            existing_server_default=esd(),
+        )
+
+    @config.requirements.identity_columns
+    @combinations(
+        ({},),
+        (dict(always=True),),
+        (dict(start=3),),
+        (dict(start=3, increment=3),),
+    )
+    def test_add_column_identity(self, kw):
+        context = op_fixture("mssql")
+        op.add_column(
+            "t1",
+            Column("some_column", Integer, sqla_compat.Identity(**kw)),
+        )
+        if "start" in kw or "increment" in kw:
+            options = "(%s,%s)" % (
+                kw.get("start", 1),
+                kw.get("increment", 1),
+            )
+        else:
+            options = ""
+        context.assert_(
+            "ALTER TABLE t1 ADD some_column INTEGER NOT NULL IDENTITY%s"
+            % options
+        )
+
+    @combinations(
+        (lambda: sqla_compat.Identity(), lambda: None),
+        (lambda: None, lambda: sqla_compat.Identity()),
+        (
+            lambda: sqla_compat.Identity(),
+            lambda: sqla_compat.Identity(),
+        ),
+    )
+    @config.requirements.identity_columns
+    def test_alter_column_identity_add_not_supported(self, sd, esd):
+        op_fixture("mssql")
+        assert_raises_message(
+            exc.CompileError,
+            'Adding, removing or modifying an "identity" construct, '
+            "e.g. GENERATED AS IDENTITY, to or from an existing "
+            "column is not supported in this dialect.",
+            op.alter_column,
+            "t1",
+            "c1",
+            server_default=sd(),
+            existing_server_default=esd(),
+        )
diff --git a/tests/test_mysql.py b/tests/test_mysql.py
index 6fcc35a..caef197 100644
--- a/tests/test_mysql.py
+++ b/tests/test_mysql.py
@@ -1,6 +1,7 @@
 from sqlalchemy import Boolean
 from sqlalchemy import Column
 from sqlalchemy import DATETIME
+from sqlalchemy import exc
 from sqlalchemy import Float
 from sqlalchemy import func
 from sqlalchemy import inspect
@@ -17,6 +18,7 @@ from alembic.autogenerate import compare
 from alembic.migration import MigrationContext
 from alembic.operations import ops
 from alembic.testing import assert_raises_message
+from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing.env import clear_staging_env
 from alembic.testing.env import staging_env
@@ -460,6 +462,52 @@ class MySQLOpTest(TestBase):
             "t1",
         )
 
+    @combinations(
+        (lambda: sqla_compat.Computed("foo * 5"), lambda: None),
+        (lambda: None, lambda: sqla_compat.Computed("foo * 5")),
+        (
+            lambda: sqla_compat.Computed("foo * 42"),
+            lambda: sqla_compat.Computed("foo * 5"),
+        ),
+    )
+    @config.requirements.computed_columns_api
+    def test_alter_column_computed_not_supported(self, sd, esd):
+        op_fixture("mssql")
+        assert_raises_message(
+            exc.CompileError,
+            'Adding or removing a "computed" construct, e.g. '
+            "GENERATED ALWAYS AS, to or from an existing column is not "
+            "supported.",
+            op.alter_column,
+            "t1",
+            "c1",
+            server_default=sd(),
+            existing_server_default=esd(),
+        )
+
+    @combinations(
+        (lambda: sqla_compat.Identity(), lambda: None),
+        (lambda: None, lambda: sqla_compat.Identity()),
+        (
+            lambda: sqla_compat.Identity(),
+            lambda: sqla_compat.Identity(),
+        ),
+    )
+    @config.requirements.identity_columns_api
+    def test_alter_column_identity_not_supported(self, sd, esd):
+        op_fixture()
+        assert_raises_message(
+            exc.CompileError,
+            'Adding, removing or modifying an "identity" construct, '
+            "e.g. GENERATED AS IDENTITY, to or from an existing "
+            "column is not supported in this dialect.",
+            op.alter_column,
+            "t1",
+            "c1",
+            server_default=sd(),
+            existing_server_default=esd(),
+        )
+
 
 class MySQLBackendOpTest(AlterColRoundTripFixture, TestBase):
     __only_on__ = "mysql", "mariadb"
@@ -578,7 +626,7 @@ class MySQLDefaultCompareTest(TestBase):
         insp = inspect(self.bind)
         cols = insp.get_columns(t1.name)
         refl = Table(t1.name, MetaData())
-        insp.reflecttable(refl, None)
+        sqla_compat._reflect_table(insp, refl, None)
         ctx = self.autogen_context["context"]
         return ctx.impl.compare_server_default(
             refl.c[cols[0]["name"]], col, rendered, cols[0]["default"]
diff --git a/tests/test_op.py b/tests/test_op.py
index 7b8dc02..58a6a86 100644
--- a/tests/test_op.py
+++ b/tests/test_op.py
@@ -19,6 +19,7 @@ from alembic import op
 from alembic.operations import ops
 from alembic.operations import schemaobj
 from alembic.testing import assert_raises_message
+from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing import eq_
 from alembic.testing import is_
@@ -374,8 +375,16 @@ class OpTest(TestBase):
         op.alter_column("t", "c", server_default=None, schema="foo")
         context.assert_("ALTER TABLE foo.t ALTER COLUMN c DROP DEFAULT")
 
+    @combinations(
+        (lambda: sqla_compat.Computed("foo * 5"), lambda: None),
+        (lambda: None, lambda: sqla_compat.Computed("foo * 5")),
+        (
+            lambda: sqla_compat.Computed("foo * 42"),
+            lambda: sqla_compat.Computed("foo * 5"),
+        ),
+    )
     @config.requirements.computed_columns_api
-    def test_alter_column_computed_add_not_supported(self):
+    def test_alter_column_computed_not_supported(self, sd, esd):
         op_fixture()
         assert_raises_message(
             exc.CompileError,
@@ -385,22 +394,31 @@ class OpTest(TestBase):
             op.alter_column,
             "t1",
             "c1",
-            server_default=sqla_compat.Computed("foo * 5"),
-        )
-
-    @config.requirements.computed_columns_api
-    def test_alter_column_computed_remove_not_supported(self):
+            server_default=sd(),
+            existing_server_default=esd(),
+        )
+
+    @combinations(
+        (lambda: sqla_compat.Identity(), lambda: None),
+        (lambda: None, lambda: sqla_compat.Identity()),
+        (
+            lambda: sqla_compat.Identity(),
+            lambda: sqla_compat.Identity(),
+        ),
+    )
+    @config.requirements.identity_columns_api
+    def test_alter_column_identity_not_supported(self, sd, esd):
         op_fixture()
         assert_raises_message(
             exc.CompileError,
-            'Adding or removing a "computed" construct, e.g. '
-            "GENERATED ALWAYS AS, to or from an existing column is not "
-            "supported.",
+            'Adding, removing or modifying an "identity" construct, '
+            "e.g. GENERATED AS IDENTITY, to or from an existing "
+            "column is not supported in this dialect.",
             op.alter_column,
             "t1",
             "c1",
-            server_default=None,
-            existing_server_default=sqla_compat.Computed("foo * 5"),
+            server_default=sd(),
+            existing_server_default=esd(),
         )
 
     def test_alter_column_schema_type_unnamed(self):
diff --git a/tests/test_oracle.py b/tests/test_oracle.py
index f706190..f84c0e1 100644
--- a/tests/test_oracle.py
+++ b/tests/test_oracle.py
@@ -1,8 +1,11 @@
 from sqlalchemy import Column
+from sqlalchemy import exc
 from sqlalchemy import Integer
 
 from alembic import command
 from alembic import op
+from alembic.testing import assert_raises_message
+from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing.env import _no_sql_testing_config
 from alembic.testing.env import clear_staging_env
@@ -68,7 +71,7 @@ class OpTest(TestBase):
             "COMMENT ON COLUMN t1.c1 IS 'c1 comment'",
         )
 
-    @config.requirements.computed_columns_api
+    @config.requirements.computed_columns
     def test_add_column_computed(self):
         context = op_fixture("oracle")
         op.add_column(
@@ -80,6 +83,29 @@ class OpTest(TestBase):
             "INTEGER GENERATED ALWAYS AS (foo * 5)"
         )
 
+    @combinations(
+        (lambda: sqla_compat.Computed("foo * 5"), lambda: None),
+        (lambda: None, lambda: sqla_compat.Computed("foo * 5")),
+        (
+            lambda: sqla_compat.Computed("foo * 42"),
+            lambda: sqla_compat.Computed("foo * 5"),
+        ),
+    )
+    @config.requirements.computed_columns
+    def test_alter_column_computed_not_supported(self, sd, esd):
+        op_fixture("oracle")
+        assert_raises_message(
+            exc.CompileError,
+            'Adding or removing a "computed" construct, e.g. '
+            "GENERATED ALWAYS AS, to or from an existing column is not "
+            "supported.",
+            op.alter_column,
+            "t1",
+            "c1",
+            server_default=sd(),
+            existing_server_default=esd(),
+        )
+
     def test_alter_table_rename_oracle(self):
         context = op_fixture("oracle")
         op.rename_table("s", "t")
@@ -226,3 +252,125 @@ class OpTest(TestBase):
     #    context.assert_(
     #        'ALTER TABLE y.t RENAME COLUMN c TO c2'
     #    )
+
+    def _identity_qualification(self, kw):
+        always = kw.get("always", False)
+        if always is None:
+            return ""
+        qualification = "ALWAYS" if always else "BY DEFAULT"
+        if kw.get("on_null", False):
+            qualification += " ON NULL"
+        return qualification
+
+    @config.requirements.identity_columns
+    @combinations(
+        ({}, None),
+        (dict(always=True), None),
+        (dict(always=None, order=True), "ORDER"),
+        (
+            dict(start=3, increment=33, maxvalue=99, cycle=True),
+            "INCREMENT BY 33 START WITH 3 MAXVALUE 99 CYCLE",
+        ),
+        (dict(on_null=True, start=42), "START WITH 42"),
+    )
+    def test_add_column_identity(self, kw, text):
+        context = op_fixture("oracle")
+        op.add_column(
+            "t1",
+            Column("some_column", Integer, sqla_compat.Identity(**kw)),
+        )
+        qualification = self._identity_qualification(kw)
+        options = " (%s)" % text if text else ""
+        context.assert_(
+            "ALTER TABLE t1 ADD some_column "
+            "INTEGER GENERATED %s AS IDENTITY%s" % (qualification, options)
+        )
+
+    @config.requirements.identity_columns
+    @combinations(
+        ({}, None),
+        (dict(always=True), None),
+        (dict(always=None, cycle=True), "CYCLE"),
+        (
+            dict(start=3, increment=33, maxvalue=99, cycle=True),
+            "INCREMENT BY 33 START WITH 3 MAXVALUE 99 CYCLE",
+        ),
+        (dict(on_null=True, start=42), "START WITH 42"),
+    )
+    def test_add_identity_to_column(self, kw, text):
+        context = op_fixture("oracle")
+        op.alter_column(
+            "t1",
+            "some_column",
+            server_default=sqla_compat.Identity(**kw),
+            existing_server_default=None,
+        )
+        qualification = self._identity_qualification(kw)
+        options = " (%s)" % text if text else ""
+        context.assert_(
+            "ALTER TABLE t1 MODIFY some_column "
+            "GENERATED %s AS IDENTITY%s" % (qualification, options)
+        )
+
+    @config.requirements.identity_columns
+    def test_remove_identity_from_column(self):
+        context = op_fixture("oracle")
+        op.alter_column(
+            "t1",
+            "some_column",
+            server_default=None,
+            existing_server_default=sqla_compat.Identity(),
+        )
+        context.assert_("ALTER TABLE t1 MODIFY some_column DROP IDENTITY")
+
+    @config.requirements.identity_columns
+    @combinations(
+        ({}, dict(always=True), None),
+        (
+            dict(always=True),
+            dict(always=False, start=3),
+            "START WITH 3",
+        ),
+        (
+            dict(always=True, start=3, increment=2, minvalue=-3, maxvalue=99),
+            dict(
+                always=True,
+                start=3,
+                increment=1,
+                minvalue=-3,
+                maxvalue=99,
+                cycle=True,
+            ),
+            "INCREMENT BY 1 START WITH 3 MINVALUE -3 MAXVALUE 99 CYCLE",
+        ),
+        (
+            dict(
+                always=False,
+                start=3,
+                maxvalue=9999,
+                minvalue=0,
+            ),
+            dict(always=False, start=3, order=True, on_null=False, cache=2),
+            "START WITH 3 CACHE 2 ORDER",
+        ),
+        (
+            dict(always=False),
+            dict(always=None, minvalue=0),
+            "MINVALUE 0",
+        ),
+    )
+    def test_change_identity_in_column(self, existing, updated, text):
+        context = op_fixture("oracle")
+        op.alter_column(
+            "t1",
+            "some_column",
+            server_default=sqla_compat.Identity(**updated),
+            existing_server_default=sqla_compat.Identity(**existing),
+        )
+
+        qualification = self._identity_qualification(updated)
+        options = " (%s)" % text if text else ""
+        context.assert_(
+            "ALTER TABLE t1 MODIFY some_column "
+            "GENERATED %s AS IDENTITY%s" % (qualification, options)
+        )
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index f928868..08f70d8 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -2,6 +2,7 @@ from sqlalchemy import BigInteger
 from sqlalchemy import Boolean
 from sqlalchemy import Column
 from sqlalchemy import DateTime
+from sqlalchemy import exc
 from sqlalchemy import Float
 from sqlalchemy import func
 from sqlalchemy import Index
@@ -37,6 +38,8 @@ from alembic.migration import MigrationContext
 from alembic.operations import Operations
 from alembic.operations import ops
 from alembic.script import ScriptDirectory
+from alembic.testing import assert_raises_message
+from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing import eq_
 from alembic.testing import eq_ignore_whitespace
@@ -276,7 +279,7 @@ class PostgresqlOpTest(TestBase):
         op.drop_table_comment("t2", existing_comment="t2 table", schema="foo")
         context.assert_("COMMENT ON TABLE foo.t2 IS NULL")
 
-    @config.requirements.computed_columns_api
+    @config.requirements.computed_columns
     def test_add_column_computed(self):
         context = op_fixture("postgresql")
         op.add_column(
@@ -288,6 +291,134 @@ class PostgresqlOpTest(TestBase):
             "INTEGER GENERATED ALWAYS AS (foo * 5) STORED"
         )
 
+    @combinations(
+        (lambda: sqla_compat.Computed("foo * 5"), lambda: None),
+        (lambda: None, lambda: sqla_compat.Computed("foo * 5")),
+        (
+            lambda: sqla_compat.Computed("foo * 42"),
+            lambda: sqla_compat.Computed("foo * 5"),
+        ),
+    )
+    @config.requirements.computed_columns
+    def test_alter_column_computed_not_supported(self, sd, esd):
+        op_fixture("postgresql")
+        assert_raises_message(
+            exc.CompileError,
+            'Adding or removing a "computed" construct, e.g. '
+            "GENERATED ALWAYS AS, to or from an existing column is not "
+            "supported.",
+            op.alter_column,
+            "t1",
+            "c1",
+            server_default=sd(),
+            existing_server_default=esd(),
+        )
+
+    @config.requirements.identity_columns
+    @combinations(
+        ({}, None),
+        (dict(always=True), None),
+        (
+            dict(start=3, increment=33, maxvalue=99, cycle=True),
+            "INCREMENT BY 33 START WITH 3 MAXVALUE 99 CYCLE",
+        ),
+    )
+    def test_add_column_identity(self, kw, text):
+        context = op_fixture("postgresql")
+        op.add_column(
+            "t1",
+            Column("some_column", Integer, sqla_compat.Identity(**kw)),
+        )
+        qualification = "ALWAYS" if kw.get("always", False) else "BY DEFAULT"
+        options = " (%s)" % text if text else ""
+        context.assert_(
+            "ALTER TABLE t1 ADD COLUMN some_column "
+            "INTEGER GENERATED %s AS IDENTITY%s" % (qualification, options)
+        )
+
+    @config.requirements.identity_columns
+    @combinations(
+        ({}, None),
+        (dict(always=True), None),
+        (
+            dict(start=3, increment=33, maxvalue=99, cycle=True),
+            "INCREMENT BY 33 START WITH 3 MAXVALUE 99 CYCLE",
+        ),
+    )
+    def test_add_identity_to_column(self, kw, text):
+        context = op_fixture("postgresql")
+        op.alter_column(
+            "t1",
+            "some_column",
+            server_default=sqla_compat.Identity(**kw),
+            existing_server_default=None,
+        )
+        qualification = "ALWAYS" if kw.get("always", False) else "BY DEFAULT"
+        options = " (%s)" % text if text else ""
+        context.assert_(
+            "ALTER TABLE t1 ALTER COLUMN some_column ADD "
+            "GENERATED %s AS IDENTITY%s" % (qualification, options)
+        )
+
+    @config.requirements.identity_columns
+    def test_remove_identity_from_column(self):
+        context = op_fixture("postgresql")
+        op.alter_column(
+            "t1",
+            "some_column",
+            server_default=None,
+            existing_server_default=sqla_compat.Identity(),
+        )
+        context.assert_(
+            "ALTER TABLE t1 ALTER COLUMN some_column DROP IDENTITY"
+        )
+
+    @config.requirements.identity_columns
+    @combinations(
+        ({}, dict(always=True), "SET GENERATED ALWAYS"),
+        (
+            dict(always=True),
+            dict(always=False, start=3),
+            "SET GENERATED BY DEFAULT SET START WITH 3",
+        ),
+        (
+            dict(always=True, start=3, increment=2, minvalue=-3, maxvalue=99),
+            dict(
+                always=True,
+                start=3,
+                increment=1,
+                minvalue=-3,
+                maxvalue=99,
+                cycle=True,
+            ),
+            "SET CYCLE SET INCREMENT BY 1",
+        ),
+        (
+            dict(
+                always=False,
+                start=3,
+                maxvalue=9999,
+                minvalue=0,
+            ),
+            dict(always=False, start=3, order=True, on_null=False, cache=2),
+            "SET CACHE 2",
+        ),
+        (
+            dict(always=False),
+            dict(always=None, minvalue=0),
+            "SET MINVALUE 0",
+        ),
+    )
+    def test_change_identity_in_column(self, existing, updated, text):
+        context = op_fixture("postgresql")
+        op.alter_column(
+            "t1",
+            "some_column",
+            server_default=sqla_compat.Identity(**updated),
+            existing_server_default=sqla_compat.Identity(**existing),
+        )
+        context.assert_("ALTER TABLE t1 ALTER COLUMN some_column %s" % text)
+
 
 class PGAutocommitBlockTest(TestBase):
     __only_on__ = "postgresql"