Commits

Mike Bayer committed de874c5

- Improved the methodology to handling percent signs in column
names from [ticket:1256]. Added more tests. MySQL and
Postgres dialects still do not issue correct CREATE TABLE
statements for identifiers with percent signs in them.

  • Participants
  • Parent commits 54c5630

Comments (0)

Files changed (5)

       combined with joined-table inheritance and an object
       which contained no defined values for the child table where
       an UPDATE with no SET clause would be rendered.
+
+- sql
+    - Improved the methodology to handling percent signs in column
+      names from [ticket:1256].  Added more tests.  MySQL and
+      Postgres dialects still do not issue correct CREATE TABLE
+      statements for identifiers with percent signs in them.
       
 - schema
     - Index now accepts column-oriented InstrumentedAttributes

lib/sqlalchemy/sql/compiler.py

                 schema_prefix = self.preparer.quote(column.table.schema, column.table.quote_schema) + '.'
             else:
                 schema_prefix = ''
-            return schema_prefix + self.preparer.quote(column.table.name % self.anon_map, column.table.quote) + "." + name
+            tablename = column.table.name
+            if isinstance(tablename, sql._generated_label):
+                tablename = tablename % self.anon_map
+            return schema_prefix + self.preparer.quote(tablename, column.table.quote) + "." + name
 
     def escape_literal_column(self, text):
         """provide escaping for the literal_column() construct."""
 
         return bind_name
 
-    _trunc_re = re.compile(r'%\((-?\d+ \w+)\)s', re.U)
     def _truncated_identifier(self, ident_class, name):
         if (ident_class, name) in self.truncated_names:
             return self.truncated_names[(ident_class, name)]
 
-        anonname = self._trunc_re.sub(lambda m: self.anon_map[m.group(1)], name)
+        anonname = name % self.anon_map 
 
         if len(anonname) > self.label_length:
             counter = self.truncated_names.get(ident_class, 1)
         return truncname
     
     def _anonymize(self, name):
-        return self._trunc_re.sub(lambda m: self.anon_map[m.group(1)], name)
+        return name % self.anon_map
         
     def _process_anon(self, key):
         (ident, derived) = key.split(' ')

lib/sqlalchemy/sql/expression.py

 class _generated_label(unicode):
     """A unicode subclass used to identify dynamically generated names."""
 
+def _escape_for_generated(x):
+    if isinstance(x, _generated_label):
+        return x
+    else:
+        return x.replace('%', '%%')
+        
 def _clone(element):
     return element._clone()
 
         expressions and function calls.
 
         """
-        return _generated_label("%%(%d %s)s" % (id(self), getattr(self, 'name', 'anon')))
+        return _generated_label("%%(%d %s)s" % (id(self), _escape_for_generated(getattr(self, 'name', 'anon'))))
 
 class ColumnCollection(util.OrderedProperties):
     """An ordered dictionary that stores a list of ColumnElement
 
         """
         if unique:
-            self.key = _generated_label("%%(%d %s)s" % (id(self), key or 'param'))
+            self.key = _generated_label("%%(%d %s)s" % (id(self), key and _escape_for_generated(key) or 'param'))
         else:
             self.key = key or _generated_label("%%(%d param)s" % id(self))
         self._orig_key = key or 'param'
     def _clone(self):
         c = ClauseElement._clone(self)
         if self.unique:
-            c.key = _generated_label("%%(%d %s)s" % (id(c), c._orig_key or 'param'))
+            c.key = _generated_label("%%(%d %s)s" % (id(c), c._orig_key and _escape_for_generated(c._orig_key) or 'param'))
         return c
 
     def _convert_to_unique(self):
         if not self.unique:
             self.unique = True
-            self.key = _generated_label("%%(%d %s)s" % (id(self), self._orig_key or 'param'))
+            self.key = _generated_label("%%(%d %s)s" % (id(self), self._orig_key and _escape_for_generated(self._orig_key) or 'param'))
 
     def bind_processor(self, dialect):
         return self.type.dialect_impl(dialect).bind_processor(dialect)
         if alias is None:
             if self.original.named_with_column:
                 alias = getattr(self.original, 'name', None)
-            alias = _generated_label('%%(%d %s)s' % (id(self), alias or 'anon'))
+            alias = _generated_label('%%(%d %s)s' % (id(self), alias and _escape_for_generated(alias) or 'anon'))
         self.name = alias
 
     @property
     def __init__(self, name, element, type_=None):
         while isinstance(element, _Label):
             element = element.element
-        self.name = self.key = self._label = name or _generated_label("%%(%d %s)s" % (id(self), getattr(element, 'name', 'anon')))
+        self.name = self.key = self._label = name or _generated_label("%%(%d %s)s" % (id(self), _escape_for_generated(getattr(element, 'name', 'anon'))))
         self._element = element
         self._type = type_
         self.quote = element.quote
 
         elif self.table and self.table.named_with_column:
             if getattr(self.table, 'schema', None):
-                label = self.table.schema + "_" + self.table.name + "_" + self.name
+                label = self.table.schema + "_" + _escape_for_generated(self.table.name) + "_" + _escape_for_generated(self.name)
             else:
-                label = self.table.name + "_" + self.name
+                label = _escape_for_generated(self.table.name) + "_" + _escape_for_generated(self.name)
 
             if label in self.table.c:
                 # TODO: coverage does not seem to be present for this

test/sql/query.py

         r = s.execute().fetchall()
         assert len(r) == 1
 
-
+class PercentSchemaNamesTest(TestBase):
+    """tests using percent signs, spaces in table and column names.
+    
+    Doesn't pass for mysql, postgres, but this is really a 
+    SQLAlchemy bug - we should be escaping out %% signs for this
+    operation the same way we do for text() and column labels.
+    
+    """
+    @testing.crashes('mysql', 'mysqldb calls name % (params)')
+    @testing.crashes('postgres', 'postgres calls name % (params)')
+    def setUpAll(self):
+        global percent_table, metadata
+        metadata = MetaData(testing.db)
+        percent_table = Table('percent%table', metadata,
+            Column("percent%", Integer),
+            Column("%(oneofthese)s", Integer),
+            Column("spaces % more spaces", Integer),
+        )
+        metadata.create_all()
+        
+    def tearDownAll(self):
+        metadata.drop_all()
+    
+    @testing.crashes('mysql', 'mysqldb calls name % (params)')
+    @testing.crashes('postgres', 'postgres calls name % (params)')
+    def test_roundtrip(self):
+        percent_table.insert().execute(
+            {'percent%':5, '%(oneofthese)s':7, 'spaces % more spaces':12},
+        )
+        percent_table.insert().execute(
+            {'percent%':7, '%(oneofthese)s':8, 'spaces % more spaces':11},
+            {'percent%':9, '%(oneofthese)s':9, 'spaces % more spaces':10},
+            {'percent%':11, '%(oneofthese)s':10, 'spaces % more spaces':9},
+        )
+        eq_(
+            percent_table.select().order_by(percent_table.c['%(oneofthese)s']).execute().fetchall(),
+            [
+                (5, 7, 12),
+                (7, 8, 11),
+                (9, 9, 10),
+                (11, 10, 9)
+            ]
+        )
+        result = percent_table.select().order_by(percent_table.c['%(oneofthese)s']).execute()
+        row = result.fetchone()
+        eq_(row[percent_table.c['percent%']], 5)
+        eq_(row[percent_table.c['%(oneofthese)s']], 7)
+        eq_(row[percent_table.c['spaces % more spaces']], 12)
+        row = result.fetchone()
+        eq_(row['percent%'], 7)
+        eq_(row['%(oneofthese)s'], 8)
+        eq_(row['spaces % more spaces'], 11)
+        result.close()
+        percent_table.update().values({percent_table.c['%(oneofthese)s']:9, percent_table.c['spaces % more spaces']:15}).execute()
+        eq_(
+            percent_table.select().order_by(percent_table.c['%(oneofthese)s']).execute().fetchall(),
+            [
+                (5, 9, 15),
+                (7, 9, 15),
+                (9, 9, 15),
+                (11, 9, 15)
+            ]
+        )
+        
+        
+        
 class LimitTest(TestBase):
 
     def setUpAll(self):

test/sql/select.py

                             "COLLATE somecol AS x")
 
     def test_percent_chars(self):
-        t = table("table",
+        t = table("table%name",
             column("percent%"),
             column("%(oneofthese)s"),
             column("spaces % more spaces"),
         )
         self.assert_compile(
             t.select(use_labels=True),
-            '''SELECT "table"."percent%" AS "table_percent%", '''\
-            '''"table"."%(oneofthese)s" AS "table_%(oneofthese)s", '''\
-            '''"table"."spaces % more spaces" AS "table_spaces % more spaces" FROM "table"'''
+            '''SELECT "table%name"."percent%" AS "table%name_percent%", '''\
+            '''"table%name"."%(oneofthese)s" AS "table%name_%(oneofthese)s", '''\
+            '''"table%name"."spaces % more spaces" AS "table%name_spaces % more spaces" FROM "table%name"'''
         )