alembic support

Author: daimor

requirements-dev.txt

@@ -3,3 +3,4 @@ flake8
 pytest
 black
 twine
+alembic

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = sqlalchemy-iris
-version = 0.9.1
+version = 0.10.0
 description = InterSystems IRIS for SQLAlchemy
 long_description = file: README.md
 url = https://github.com/caretdev/sqlalchemy-iris

sqlalchemy_iris/__init__.py

@@ -3,6 +3,13 @@
 from . import base
 from . import iris
 
+try:
+    import alembic
+except ImportError:
+    pass
+else:
+    from .alembic import IRISImpl
+
 from .base import BIGINT
 from .base import BIT
 from .base import DATE

sqlalchemy_iris/alembic.py

@@ -0,0 +1,105 @@
+import logging
+
+from sqlalchemy.ext.compiler import compiles
+from alembic.ddl import DefaultImpl
+from alembic.ddl.base import ColumnNullable
+from alembic.ddl.base import ColumnType
+from alembic.ddl.base import ColumnName
+from alembic.ddl.base import Column
+from alembic.ddl.base import alter_table
+from alembic.ddl.base import alter_column
+from alembic.ddl.base import format_type
+from alembic.ddl.base import format_column_name
+
+from .base import IRISDDLCompiler
+
+log = logging.getLogger(__name__)
+
+
+class IRISImpl(DefaultImpl):
+    __dialect__ = "iris"
+
+    type_synonyms = DefaultImpl.type_synonyms + (
+        {"BLOB", "LONGVARBINARY"},
+        {"DOUBLE", "FLOAT"},
+        {"DATETIME", "TIMESTAMP"},
+    )
+
+    def compare_type(self, inspector_column: Column, metadata_column: Column) -> bool:
+        # Don't change type of IDENTITY column
+        if (
+            metadata_column.primary_key
+            and metadata_column is metadata_column.table._autoincrement_column
+        ):
+            return False
+
+        return super().compare_type(inspector_column, metadata_column)
+
+    def compare_server_default(
+        self,
+        inspector_column: Column,
+        metadata_column: Column,
+        rendered_metadata_default,
+        rendered_inspector_default,
+    ):
+        # don't do defaults for IDENTITY columns
+        if (
+            metadata_column.primary_key
+            and metadata_column is metadata_column.table._autoincrement_column
+        ):
+            return False
+
+        return super().compare_server_default(
+            inspector_column,
+            metadata_column,
+            rendered_metadata_default,
+            rendered_inspector_default,
+        )
+
+    def correct_for_autogen_constraints(
+        self,
+        conn_unique_constraints,
+        conn_indexes,
+        metadata_unique_constraints,
+        metadata_indexes,
+    ):
+
+        doubled_constraints = {
+            index
+            for index in conn_indexes
+            if index.info.get("duplicates_constraint")
+        }
+
+        for ix in doubled_constraints:
+            conn_indexes.remove(ix)
+
+        # if not sqla_compat.sqla_2:
+        #     self._skip_functional_indexes(metadata_indexes, conn_indexes)
+
+@compiles(ColumnNullable, "iris")
+def visit_column_nullable(
+    element: ColumnNullable, compiler: IRISDDLCompiler, **kw
+) -> str:
+    return "%s %s %s" % (
+        alter_table(compiler, element.table_name, element.schema),
+        alter_column(compiler, element.column_name),
+        "NULL" if element.nullable else "NOT NULL",
+    )
+
+
+@compiles(ColumnType, "iris")
+def visit_column_type(element: ColumnType, compiler: IRISDDLCompiler, **kw) -> str:
+    return "%s %s %s" % (
+        alter_table(compiler, element.table_name, element.schema),
+        alter_column(compiler, element.column_name),
+        "%s" % format_type(compiler, element.type_),
+    )
+
+
+@compiles(ColumnName, "iris")
+def visit_rename_column(element: ColumnName, compiler: IRISDDLCompiler, **kw) -> str:
+    return "%s %s RENAME %s" % (
+        alter_table(compiler, element.table_name, element.schema),
+        alter_column(compiler, element.column_name),
+        format_column_name(compiler, element.newname),
+    )

sqlalchemy_iris/base.py

@@ -63,8 +63,10 @@ def unique_constraints(cls):
         def check_constraints(cls):
             return []
 
+
 from sqlalchemy.types import BIGINT
 from sqlalchemy.types import VARCHAR
+from sqlalchemy.types import CHAR
 from sqlalchemy.types import INTEGER
 from sqlalchemy.types import DATE
 from sqlalchemy.types import TIMESTAMP
@@ -87,6 +89,7 @@ def check_constraints(cls):
 from .types import IRISDate
 from .types import IRISDateTime
 
+
 ischema_names = {
     "BIGINT": BIGINT,
     "BIT": BIT,
@@ -634,7 +637,7 @@ def visit_computed_column(self, generated, **kwargs):
         text = self.sql_compiler.process(
             generated.sqltext, include_table=True, literal_binds=True
         )
-        text = re.sub(r"(?<!')(\b[^\d]\w+\b)", r"{\g<1>}", text)
+        text = re.sub(r"(?<!')(\b[^\W\d]+\w+\b)", r"{\g<1>}", text)
         # text = text.replace("'", '"')
         text = "COMPUTECODE {Set {*} = %s}" % (text,)
         if generated.persisted is False:
@@ -722,6 +725,12 @@ def visit_BIT(self, type_, **kw):
     def visit_TEXT(self, type_, **kw):
         return "VARCHAR(65535)"
 
+    def visit_LONGVARBINARY(self, type_, **kw):
+        return "LONGVARBINARY"
+
+    def visit_DOUBLE(self, type_, **kw):
+        return "DOUBLE"
+
 
 class IRISIdentifierPreparer(sql.compiler.IdentifierPreparer):
     """Install IRIS specific reserved words."""
@@ -1428,22 +1437,20 @@ def get_multi_foreign_keys(
             )
         )
 
-        rs = connection.execute(s)
+        rs = connection.execution_options(future_result=True).execute(s)
 
         fkeys = util.defaultdict(dict)
 
-        for row in rs:
-            (
-                table_name,
-                rfknm,
-                scol,
-                rschema,
-                rtbl,
-                rcol,
-                _,  # match rule
-                fkuprule,
-                fkdelrule,
-            ) = row
+        for row in rs.mappings():
+            table_name = row[key_constraints.c.table_name]
+            rfknm = row[key_constraints.c.constraint_name]
+            scol = row[key_constraints.c.column_name]
+            rschema = row[key_constraints_ref.c.table_schema]
+            rtbl = row[key_constraints_ref.c.table_name]
+            rcol = row[key_constraints_ref.c.column_name]
+            _ = row[ref_constraints.c.match_option]
+            fkuprule = row[ref_constraints.c.update_rule]
+            fkdelrule = row[ref_constraints.c.delete_rule]
 
             table_fkey = fkeys[(schema, table_name)]
 
@@ -1540,7 +1547,10 @@ def get_multi_columns(
             ).outerjoin(
                 property,
                 sql.and_(
-                    property.c.SqlFieldName == columns.c.column_name,
+                    sql.or_(
+                        property.c.Name == columns.c.column_name,
+                        property.c.SqlFieldName == columns.c.column_name,
+                    ),
                     property.c.parent
                     == sql.select(tables.c.classname)
                     .where(
@@ -1593,6 +1603,9 @@ def get_multi_columns(
             if coltype is None:
                 util.warn("Did not recognize type '%s' of column '%s'" % (type_, name))
                 coltype = sqltypes.NULLTYPE
+            elif coltype is VARCHAR and charlen == 1:
+                # VARCHAR(1) as CHAR
+                coltype = CHAR
             else:
                 if issubclass(coltype, sqltypes.Numeric):
                     kwargs["precision"] = int(numericprec)
@@ -1602,6 +1615,10 @@ def get_multi_columns(
 
                 coltype = coltype(**kwargs)
 
+            default = "" if default == "$c(0)" else default
+            if default and default.startswith('"'):
+                default = "'%s'" % (default[1:-1].replace("'", "''"),)
+
             cdict = {
                 "name": name,
                 "type": coltype,

sqlalchemy_iris/information_schema.py

@@ -76,6 +76,7 @@ def process_result_value(self, value, dialect):
     "PropertyDefinition",
     ischema,
     Column("parent", String),
+    Column("Name", String),
     Column("SqlFieldName", String),
     Column("SqlComputeCode", String),
     Column("SqlComputed", Boolean),

sqlalchemy_iris/requirements.py

@@ -1,9 +1,18 @@
 from sqlalchemy.testing.requirements import SuiteRequirements
 
+try:
+    from alembic.testing.requirements import SuiteRequirements as AlembicRequirements
+except:  # noqa
+    from sqlalchemy.testing.requirements import Requirements as BaseRequirements
+
+    class AlembicRequirements(BaseRequirements):
+        pass
+
+
 from sqlalchemy.testing import exclusions
 
 
-class Requirements(SuiteRequirements):
+class Requirements(SuiteRequirements, AlembicRequirements):
     @property
     def array_type(self):
         return exclusions.closed()
@@ -121,7 +130,6 @@ def foreign_key_constraint_option_reflection_onupdate(self):
     def fk_constraint_option_reflection_onupdate_restrict(self):
         return exclusions.closed()
 
-
     @property
     def precision_numerics_many_significant_digits(self):
         """target backend supports values with many digits on both sides,
@@ -239,3 +247,13 @@ def unique_index_reflect_as_unique_constraints(self):
         """Target database reflects unique indexes as unique constrains."""
 
         return exclusions.open()
+
+    # alembic
+
+    @property
+    def fk_onupdate_restrict(self):
+        return exclusions.closed()
+
+    @property
+    def fk_ondelete_restrict(self):
+        return exclusions.closed()

tests/test_alembic.py

@@ -0,0 +1,25 @@
+try:
+    import alembic  # noqa
+except:  # noqa
+    pass
+else:
+    from alembic.testing.suite.test_op import (
+        BackendAlterColumnTest as _BackendAlterColumnTest,
+    )
+    from alembic.testing.suite.test_autogen_diffs import (
+        AutoincrementTest as _AutoincrementTest,
+    )
+    from alembic.testing.suite import *  # noqa
+
+    class BackendAlterColumnTest(_BackendAlterColumnTest):
+        def test_rename_column(self):
+            # IRIS Uppercases new names
+            self._run_alter_col({}, {"name": "NEWNAME"})
+
+    class AutoincrementTest(_AutoincrementTest):
+        # pk don't change type
+        def test_alter_column_autoincrement_pk_implicit_true(self):
+            pass
+
+        def test_alter_column_autoincrement_pk_explicit_true(self):
+            pass