I am trying to re-write this mysql query in SQLAlchemy:
Schema:
CREATE TABLE `posts` ( `post_id` INT UNSIGNED PRIMARY AUTO_INCREMENT, `post_name` VARCHAR(255) ) Engine=InnoDB; CREATE TABLE `post_tags` ( `tag_id` INT UNSIGNED PRIMARY AUTO_INCREMENT, `tag_name` VARCHAR(255) ) Engine=InnoDB; CREATE TABLE `post_tags_map` ( `map_id` INT PRIMARY AUTO_INCREMENT, `post_id` INT NOT NULL, `tags_id` INT NOT NULL, FOREIGN KEY `post_id` REFERENCES `posts` (`post_id`), FOREIGN KEY `post_id` REFERENCES `post_tags` (`tag_id`) ) Engine=InnoDB;
Query:
SELECT posts.*, GROUP_CONCAT( post_tags.tag_name order by post_tags.tag_name ) AS tags FROM posts LEFT JOIN posts_tags_map ON posts_tags_map.post_id = posts.post_id LEFT JOIN post_tags ON posts_tags_map.tags_id = posts_tags.tag_id WHERE posts.post_id = 1 GROUP BY post_id
Here’s what I have, but I keep getting:
1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'SEPARATOR /)) rows = DBSession.query(Posts,func.group_concat(Post_Tags.tag_name.op('SEPARATOR')(literal_column('/')))).outerjoin(PostsTagsMap,Posts.post_id==PostsTagsMap.post_id).outerjoin(Post_Tags,PostsTagsMap.tags_id==Post_Tags.tag_id).group_by(Posts.post_id)
Advertisement
Answer
Defining a custom function element seems the easiest way to take care of the group_concat
method.
from sqlalchemy.sql import expression import sqlalchemy from sqlalchemy.ext import compiler class group_concat(expression.FunctionElement): name = "group_concat" @compiler.compiles(group_concat, 'mysql') def _group_concat_mysql(element, compiler, **kw): if len(element.clauses) == 2: separator = compiler.process(element.clauses.clauses[1]) else: separator = ',' return 'GROUP_CONCAT(%s SEPARATOR %s)'.format( compiler.process(element.clauses.clauses[0]), separator, )
And use it something like this:
query = select([ table.c.some_column, expression.label( 'grouped column', group_concat( table.c.some_oter_column, ' separator ', ), ), ]).group_by(table.c.some_column)