fixed concat function

Author: daimor

sqlalchemy_iris/base.py

@@ -591,6 +591,12 @@ def visit_concat_op_binary(self, binary, operator, **kw):
             self.process(binary.right, **kw),
         )
 
+    def visit_concat_func(
+        self, func, **kw
+    ):
+        args = [self.process(clause, **kw) for clause in func.clauses.clauses]
+        return ' || '.join(args)
+
     def visit_mod_binary(self, binary, operator, **kw):
         return (
             self.process(binary.left, **kw) + " # " + self.process(binary.right, **kw)
@@ -955,6 +961,11 @@ def _set_option(self, connection, option, value):
                 return row[0]
         return None
 
+    def get_isolation_level_values(self, dbapi_connection):
+        levels = set(self._isolation_lookup)
+        levels.add("AUTOCOMMIT")
+        return levels
+
     def get_isolation_level(self, connection):
         try:
             level = int(self._get_option(connection, "IsolationMode"))

tests/test_suite.py

@@ -9,10 +9,11 @@
 from sqlalchemy.testing import config
 from sqlalchemy.orm import Session
 from sqlalchemy import testing
-from sqlalchemy import Table, Column, select
+from sqlalchemy import Table, Column, select, func
 from sqlalchemy.types import Integer
 from sqlalchemy.types import String
 from sqlalchemy.types import VARBINARY
+from sqlalchemy.types import TEXT
 from sqlalchemy.types import BINARY
 from sqlalchemy_iris import TINYINT
 from sqlalchemy_iris import INTEGER
@@ -440,3 +441,48 @@ def test_max_inner_product(self):
                 (2,),
             ],
         )
+
+
+class ConcatTest(fixtures.TablesTest):
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "data",
+            metadata,
+            Column("sometext", TEXT),
+            Column("testdata", TEXT),
+        )
+
+    @classmethod
+    def fixtures(cls):
+        return dict(
+            data=(
+                (
+                    "sometext",
+                    "testdata",
+                ),
+                (
+                    "sometestdata",
+                    "test",
+                ),
+            )
+        )
+
+    def _assert_result(self, select, result):
+        with config.db.connect() as conn:
+            eq_(conn.execute(select).fetchall(), result)
+
+    def test_concat_func(self):
+        self._assert_result(
+            select(
+                self.tables.data.c.sometext,
+            ).filter(
+                self.tables.data.c.sometext
+                == func.concat("some", self.tables.data.c.testdata, "data")
+            ),
+            [
+                ("sometestdata",),
+            ],
+        )