Commits

Mike Bayer committed b6838a6

Fixed bug where keyword arguments passed to
:meth:`.Compiler.process` wouldn't get propagated
to the column expressions present in the columns
clause of a SELECT statement. In particular this would
come up when used by custom compilation schemes that
relied upon special flags. [ticket:2593]

Comments (0)

Files changed (4)

doc/build/changelog/changelog_08.rst

     :released:
 
     .. change::
+        :tags: sql, bug
+        :tickets: 2593
+
+        Fixed bug where keyword arguments passed to
+        :meth:`.Compiler.process` wouldn't get propagated
+        to the column expressions present in the columns
+        clause of a SELECT statement.  In particular this would
+        come up when used by custom compilation schemes that
+        relied upon special flags.
+
+    .. change::
         :tags: sql, feature
 
       Added a new method :meth:`.Engine.execution_options`

lib/sqlalchemy/engine/interfaces.py

     defaults.
     """
 
-    def __init__(self, dialect, statement, bind=None):
+    def __init__(self, dialect, statement, bind=None,
+                compile_kwargs=util.immutabledict()):
         """Construct a new ``Compiled`` object.
 
         :param dialect: ``Dialect`` to compile against.
 
         :param bind: Optional Engine or Connection to compile this
           statement against.
+
+        :param compile_kwargs: additional kwargs that will be
+         passed to the initial call to :meth:`.Compiled.process`.
         """
 
         self.dialect = dialect
         if statement is not None:
             self.statement = statement
             self.can_execute = statement.supports_execution
-            self.string = self.process(self.statement)
+            self.string = self.process(self.statement, **compile_kwargs)
 
     @util.deprecated("0.7", ":class:`.Compiled` objects now compile "
                         "within the constructor.")

lib/sqlalchemy/sql/compiler.py

         else:
             result_expr = col_expr
 
+        column_clause_args.update(
+                    within_columns_clause=within_columns_clause,
+                    add_to_result_map=add_to_result_map
+                )
         return result_expr._compiler_dispatch(
-                       self, within_columns_clause=within_columns_clause,
-                        add_to_result_map=add_to_result_map,
+                       self,
                         **column_clause_args
                     )
 
         self.stack.append({'from': correlate_froms,
                             'iswrapper': iswrapper})
 
-        column_clause_args = {'positional_names': positional_names}
+        column_clause_args = kwargs.copy()
+        column_clause_args.update({
+                'positional_names': positional_names,
+                'within_label_clause': False,
+                'within_columns_clause': False
+            })
 
         # the actual list of columns to print in the SELECT column list.
         inner_columns = [

test/sql/test_compiler.py

     insert, literal, and_, null, type_coerce, alias, or_, literal_column,\
     Float, TIMESTAMP, Numeric, Date, Text, collate, union, except_,\
     intersect, union_all, Boolean, distinct, join, outerjoin, asc, desc,\
-    over, subquery
+    over, subquery, case
 import decimal
 from sqlalchemy import exc, sql, util, types, schema
 from sqlalchemy.sql import table, column, label
         )
 
 
+class KwargPropagationTest(fixtures.TestBase):
+
+    @classmethod
+    def setup_class(cls):
+        from sqlalchemy.sql.expression import ColumnClause, TableClause
+        class CatchCol(ColumnClause):
+            pass
+
+        class CatchTable(TableClause):
+            pass
+
+        cls.column = CatchCol("x")
+        cls.table = CatchTable("y")
+        cls.criterion = cls.column == CatchCol('y')
+
+        @compiles(CatchCol)
+        def compile_col(element, compiler, **kw):
+            assert "canary" in kw
+            return compiler.visit_column(element)
+
+        @compiles(CatchTable)
+        def compile_table(element, compiler, **kw):
+            assert "canary" in kw
+            return compiler.visit_table(element)
+
+    def _do_test(self, element):
+        d = default.DefaultDialect()
+        d.statement_compiler(d, element,
+                        compile_kwargs={"canary": True})
+
+    def test_binary(self):
+        self._do_test(self.column == 5)
+
+    def test_select(self):
+        s = select([self.column]).select_from(self.table).\
+                where(self.column == self.criterion).\
+                order_by(self.column)
+        self._do_test(s)
+
+    def test_case(self):
+        c = case([(self.criterion, self.column)], else_=self.column)
+        self._do_test(c)
+
+    def test_cast(self):
+        c = cast(self.column, Integer)
+        self._do_test(c)
+
 class CRUDTest(fixtures.TestBase, AssertsCompiledSQL):
     __dialect__ = 'default'