small updates

Author: daimor

setup.cfg

@@ -14,9 +14,11 @@ classifiers =
     License :: OSI Approved :: MIT License
     Programming Language :: Python
     Programming Language :: Python :: 3
-    Programming Language :: Python :: 3.8
-    Programming Language :: Python :: 3.9
     Programming Language :: Python :: 3.10
+    Programming Language :: Python :: 3.11
+    Programming Language :: Python :: 3.12
+    Programming Language :: Python :: 3.13
+    Programming Language :: Python :: 3.14
     Topic :: Database :: Front-Ends
     Operating System :: OS Independent
 keywords="SQLAlchemy InterSystems IRIS",
@@ -25,12 +27,12 @@ project_urls =
     Tracker = https://github.com/caretdev/sqlalchemy-iris/issues
 
 [options]
-python_requires = >=3.8
+python_requires = >=3.10
 packages = find:
 
 [options.extras_require]
 intersystems =
-    intersystems-irispython==5.1.0
+    intersystems-irispython~=5.1.0
 
 [tool:pytest]
 addopts= --tb native -v -r fxX -p no:warnings

sqlalchemy_iris/base.py

@@ -1,5 +1,7 @@
 import re
 from decimal import Decimal
+from typing import Optional
+from typing import Any
 import intersystems_iris.dbapi._DBAPI as dbapi
 from . import information_schema as ischema
 from sqlalchemy import exc
@@ -110,6 +112,7 @@ def check_constraints(cls):
     "TINYINT": TINYINT,
     "VARBINARY": VARBINARY,
     "VARCHAR": VARCHAR,
+    "CHAR": CHAR,
 }
 
 RESERVED_WORDS = set(
@@ -418,7 +421,10 @@ def limit_clause(self, select, **kw):
 
             if limit_clause is not None:
                 sql += " LIMIT %s" % self.process(limit_clause, **kw)
-            elif offset_clause is not None and self.dialect.server_version_info < (2025, 2):
+            elif offset_clause is not None and self.dialect.server_version_info < (
+                2025,
+                2,
+            ):
                 # IRIS 2025.1 has a bug with offset without limit
                 sql += " LIMIT 100"
             if offset_clause is not None:
@@ -499,7 +505,11 @@ def get_select_precolumns(self, select, **kw):
             else:
                 text += "DISTINCT "
 
-        if not self.dialect.supports_modern_pagination and select._has_row_limiting_clause and self._use_top(select):
+        if (
+            not self.dialect.supports_modern_pagination
+            and select._has_row_limiting_clause
+            and self._use_top(select)
+        ):
             text += "TOP %s " % self.process(self._get_limit_or_fetch(select), **kw)
 
         return text
@@ -635,7 +645,12 @@ def order_by_clause(self, select, **kw):
         order_by = self.process(select._order_by_clause, **kw)
         limit_clause = self._get_limit_or_fetch(select)
 
-        if order_by and not self.is_subquery() and limit_clause is None and select._offset_clause is None:
+        if (
+            order_by
+            and not self.is_subquery()
+            and limit_clause is None
+            and select._offset_clause is None
+        ):
             return " ORDER BY " + order_by
         return ""
 
@@ -645,11 +660,9 @@ def visit_concat_op_binary(self, binary, operator, **kw):
             self.process(binary.right, **kw),
         )
 
-    def visit_concat_func(
-        self, func, **kw
-    ):
+    def visit_concat_func(self, func, **kw):
         args = [self.process(clause, **kw) for clause in func.clauses.clauses]
-        return ' || '.join(args)
+        return " || ".join(args)
 
     def visit_mod_binary(self, binary, operator, **kw):
         return (
@@ -712,7 +725,7 @@ def create_table_constraints(self, table, **kw):
         comment = table.comment
         if comment:
             # hack to keep \r, kind of
-            comment = comment.replace('\r', '\n\t')
+            comment = comment.replace("\r", "\n\t")
             literal = self.sql_compiler.render_literal_value(comment, sqltypes.String())
             description = "%DESCRIPTION " + literal
 
@@ -769,7 +782,7 @@ def get_column_specification(self, column, **kwargs):
 
         comment = column.comment
         if comment is not None:
-            comment = comment.replace('\r', '\n\t')
+            comment = comment.replace("\r", "\n\t")
             literal = self.sql_compiler.render_literal_value(comment, sqltypes.String())
             colspec.append("%DESCRIPTION " + literal)
 
@@ -816,11 +829,37 @@ def visit_BOOLEAN(self, type_, **kw):
     def visit_BIT(self, type_, **kw):
         return "BIT"
 
-    def visit_VARCHAR(self, type_, **kw):
-        # If length is not specified, use 50 as default in IRIS
-        if type_.length is None:
-            type_ = VARCHAR(50)
-        return "VARCHAR(%d)" % type_.length
+    # def visit_VARCHAR(self, type_, **kw):
+    #     # If length is not specified, use 50 as default in IRIS
+    #     if type_.length is None:
+    #         type_ = VARCHAR(50)
+    #     return "VARCHAR(%d)" % type_.length
+
+    def _render_string_type(
+        self, name: str, length: Optional[int], collation: Optional[str]
+    ) -> str:
+        text = name
+        if length:
+            text += f"({length})"
+        # if collation:
+        #     text += f' COLLATE "{collation}"'
+        return text
+
+    def visit_CHAR(self, type_: sqltypes.CHAR, **kw: Any) -> str:
+        return self._render_string_type("CHAR", type_.length, type_.collation)
+
+    def visit_NCHAR(self, type_: sqltypes.NCHAR, **kw: Any) -> str:
+        return self._render_string_type("NCHAR", type_.length, type_.collation)
+
+    def visit_VARCHAR(self, type_: sqltypes.String, **kw: Any) -> str:
+        return self._render_string_type(
+            "VARCHAR", type_.length, type_.collation
+        )
+
+    def visit_NVARCHAR(self, type_: sqltypes.NVARCHAR, **kw: Any) -> str:
+        return self._render_string_type(
+            "NVARCHAR", type_.length, type_.collation
+        )
 
     def visit_TEXT(self, type_, **kw):
         return "VARCHAR(65535)"
@@ -1015,7 +1054,9 @@ def on_connect(conn):
                 self.supports_vectors = False
             self._dictionary_access = False
             with conn.cursor() as cursor:
-                res = cursor.execute("%CHECKPRIV SELECT ON %Dictionary.PropertyDefinition")
+                res = cursor.execute(
+                    "%CHECKPRIV SELECT ON %Dictionary.PropertyDefinition"
+                )
                 self._dictionary_access = res == 0
 
             # if not self.supports_vectors:
@@ -1034,7 +1075,7 @@ def on_connect(conn):
 
     def _get_option(self, connection, option):
         with connection.cursor() as cursor:
-            cursor.execute("SELECT %SYSTEM_SQL.Util_GetOption(?)", (option, ))
+            cursor.execute("SELECT %SYSTEM_SQL.Util_GetOption(?)", (option,))
             row = cursor.fetchone()
             return row[0] if row else None
 
@@ -1182,15 +1223,19 @@ def get_schema(self, schema=None):
 
     @reflection.cache
     def get_table_options(self, connection, table_name, schema=None, **kw):
-        if not self.has_table(connection=connection, table_name=table_name, schema=schema):
+        if not self.has_table(
+            connection=connection, table_name=table_name, schema=schema
+        ):
             raise exc.NoSuchTableError(
                 f"{schema}.{table_name}" if schema else table_name
             ) from None
         return {}
 
     @reflection.cache
     def get_table_comment(self, connection, table_name, schema=None, **kw):
-        if not self.has_table(connection=connection, table_name=table_name, schema=schema):
+        if not self.has_table(
+            connection=connection, table_name=table_name, schema=schema
+        ):
             raise exc.NoSuchTableError(
                 f"{schema}.{table_name}" if schema else table_name
             ) from None
@@ -1207,9 +1252,9 @@ def get_table_comment(self, connection, table_name, schema=None, **kw):
         comment = connection.execute(s).scalar()
         if comment:
             # make it as \r
-            comment = comment.replace(' \t\t\t\t', '\r')
+            comment = comment.replace(" \t\t\t\t", "\r")
             # restore \n
-            comment = comment.replace(' \t\t\t', '\n')
+            comment = comment.replace(" \t\t\t", "\n")
         return {"text": comment}
 
     @reflection.cache
@@ -1761,15 +1806,16 @@ def get_multi_columns(
             comment = row[columns.c.description]
             if comment:
                 # make it as \r
-                comment = comment.replace(' \t\t\t\t', '\r')
+                comment = comment.replace(" \t\t\t\t", "\r")
                 # restore \n
-                comment = comment.replace(' \t\t\t', '\n')
+                comment = comment.replace(" \t\t\t", "\n")
 
             coltype = self.ischema_names.get(type_, None)
 
             kwargs = {}
             if coltype in (
                 VARCHAR,
+                CHAR,
                 BINARY,
                 TEXT,
                 VARBINARY,

tests/test_suite.py

@@ -7,12 +7,16 @@
 from sqlalchemy.testing.suite import CTETest as _CTETest
 from sqlalchemy.testing.suite import DifficultParametersTest as _DifficultParametersTest
 from sqlalchemy.testing.suite import ComponentReflectionTest as _ComponentReflectionTest
+from sqlalchemy.testing.suite import (
+    ComponentReflectionTestExtra as _ComponentReflectionTestExtra,
+)
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.assertions import eq_
 from sqlalchemy.testing import config
 from sqlalchemy.orm import Session
 from sqlalchemy import testing
 from sqlalchemy import Table, Column, select, func, text
+from sqlalchemy import types as sql_types
 from sqlalchemy.types import Integer
 from sqlalchemy.types import String
 from sqlalchemy.types import VARBINARY
@@ -32,28 +36,56 @@
 
 if sqlalchemy_version.startswith("2."):
     from sqlalchemy.testing.suite import (
-        BizarroCharacterFKResolutionTest as _BizarroCharacterFKResolutionTest,
+        BizarroCharacterTest as _BizarroCharacterTest,
     )
 
-    class BizarroCharacterFKResolutionTest(_BizarroCharacterFKResolutionTest):
-        @testing.combinations(
-            ("id",), ("(3)",), ("col%p",), ("[brack]",), argnames="columnname"
-        )
+    class BizarroCharacterTest(_BizarroCharacterTest):
+        def column_names():
+            return testing.combinations(
+                ("plainname",),
+                ("(3)",),
+                ("col%p",),
+                ("[brack]",),
+                argnames="columnname",
+            )
+
+        def table_names():
+            return testing.combinations(
+                ("plain",),
+                # ("(2)",),
+                ("per % cent",),
+                ("[brackets]",),
+                argnames="tablename",
+            )
+
         @testing.variation("use_composite", [True, False])
-        @testing.combinations(
-            ("plain",),
-            # ("(2)",), not in IRIS
-            ("per % cent",),
-            ("[brackets]",),
-            argnames="tablename",
-        )
+        @column_names()
+        @table_names()
+        @testing.requires.foreign_key_constraint_reflection
         def test_fk_ref(
             self, connection, metadata, use_composite, tablename, columnname
         ):
             super().test_fk_ref(
                 connection, metadata, use_composite, tablename, columnname
             )
 
+        @column_names()
+        @table_names()
+        @testing.requires.comment_reflection
+        def test_reflect_comments(self, tablename, columnname, connection, metadata):
+            super().test_reflect_comments(tablename, columnname, connection, metadata)
+
+    class ComponentReflectionTestExtra(_ComponentReflectionTestExtra):
+
+        @testing.requires.table_reflection
+        @testing.combinations(
+            sql_types.String,
+            sql_types.VARCHAR,
+            argnames="type_",
+        )
+        def test_string_length_reflection(self, connection, metadata, type_):
+            super().test_string_length_reflection(connection, metadata, type_, [])
+
 
 class CompoundSelectTest(_CompoundSelectTest):
     @pytest.mark.skip()

tox.ini

@@ -1,36 +1,37 @@
 [tox]
 requires =
   tox>=4
-env_list = py{310,311,312}-{old,new}-{iris,intersystems}
+env_list = py{310,311,312,313,314}-{old,new}-{iris,intersystems}
 
+[testenv]
+setenv =
+    SQLALCHEMY_SILENCE_UBER_WARNING=1
 
-[testenv:py{38,39,310,311,312}-old-iris]
+[testenv:py{310,311,312,313,314}-old-iris]
 deps =
   sqlalchemy<2
   -r requirements-dev.txt
   -r requirements-iris.txt
   -e.
 commands = {envpython} -m pytest {posargs} --driver iris
-setenv =
-    SQLALCHEMY_SILENCE_UBER_WARNING = 1
 
-[testenv:py{38,39,310,311,312}-new-iris]
+[testenv:py{310,311,312,313,314}-new-iris]
 deps =
   sqlalchemy>=2
   -r requirements-dev.txt
   -r requirements-iris.txt
   -e.
 commands = {envpython} -m pytest {posargs} --driver iris
 
-[testenv:py{38,39,310,311,312}-old-intersystems]
+[testenv:py{310,311,312,313,314}-old-intersystems]
 deps =
   sqlalchemy<2
   -r requirements-dev.txt
   -r requirements-iris.txt
   -e .[intersystems]
 commands = {envpython} -m pytest {posargs} --driver intersystems
 
-[testenv:py{38,39,310,311,312}-new-intersystems]
+[testenv:py{310,311,312,313,314}-new-intersystems]
 deps =
   sqlalchemy>=2
   -r requirements-dev.txt