Add support for `string_agg` with MySQL backend

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 0a68b3779e7b4fe1636add9bb789b96987077016..4712c9b40226c2cfab04a89e07cfff20ba2f88ab 100644 (file)
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1208,6 +1208,15 @@ class MySQLCompiler(compiler.SQLCompiler):
         )
         return f"{clause} WITH ROLLUP"
 
+    def visit_string_agg_func(self, fn, **kw):
+        if len(fn.clauses) > 1:
+            return "group_concat(%s SEPARATOR %s)" % (
+                fn.clauses[0]._compiler_dispatch(self, **kw),
+                fn.clauses[-1]._compiler_dispatch(self, **kw),
+            )
+        else:
+            return "group_concat%s" % self.function_argspec(fn)
+
     def visit_sequence(self, seq, **kw):
         return "nextval(%s)" % self.preparer.format_sequence(seq)
 

diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index ba074db80c6ec11de4031f8f5ca0452883622387..ad5fcde1ae90d42b4729a3bb68c59ce955e4b523 100644 (file)
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -2766,6 +2766,9 @@ class ClauseList(
         self._is_implicitly_boolean = False
         return self
 
+    def __getitem__(self, item) -> ColumnElement[Any]:
+        return self.clauses[item]
+
     def __iter__(self) -> Iterator[ColumnElement[Any]]:
         return iter(self.clauses)
 

diff --git a/test/sql/test_functions.py b/test/sql/test_functions.py
index 51ed638940c5bbfa0f2609ef9d617b135522b64a..c15b7638884a27fb9ff9b010215ad1d90f763b5e 100644 (file)
--- a/test/sql/test_functions.py
+++ b/test/sql/test_functions.py
@@ -242,6 +242,14 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             literal_binds=True,
             render_postcompile=True,
         )
+        self.assert_compile(
+            stmt,
+            "SELECT group_concat(t.value SEPARATOR ',') "
+            "AS string_agg_1 FROM t",
+            dialect=mysql.dialect(),
+            literal_binds=True,
+            render_postcompile=True,
+        )
 
     def test_cube_operators(self):
         t = table(