- 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.

diff --git a/CHANGES b/CHANGES
index c7846b9c0d916fe52ec893eeeefa17f895275a7d..56f2e72ab8db6507cb1be2c9c45d922eb2bed427 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -45,6 +45,12 @@ CHANGES
       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

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index d5c85d71d62ec597edbae5de4bb5cb7c21775c07..8622aeea41f14a046ba0701e6536cd120175b943 100644 (file)
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -277,7 +277,10 @@ class DefaultCompiler(engine.Compiled):
                 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."""
@@ -410,12 +413,11 @@ class DefaultCompiler(engine.Compiled):
 
         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)
@@ -427,7 +429,7 @@ class DefaultCompiler(engine.Compiled):
         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(' ')

diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index f527c6351f7e580d2ae46b530fac382b0e1c8aa0..48b1d0687fc2409eaa5529baddf05bbec97c6b82 100644 (file)
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -869,6 +869,12 @@ modifier = _FunctionGenerator(group=False)
 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()
 
@@ -1637,7 +1643,7 @@ class ColumnElement(ClauseElement, _CompareMixin):
         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
@@ -1969,7 +1975,7 @@ class _BindParamClause(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'
@@ -1988,13 +1994,13 @@ class _BindParamClause(ColumnElement):
     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)
@@ -2601,7 +2607,7 @@ class Alias(FromClause):
         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
@@ -2722,7 +2728,7 @@ class _Label(ColumnElement):
     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
@@ -2811,9 +2817,9 @@ class ColumnClause(_Immutable, ColumnElement):
 
         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

diff --git a/test/sql/query.py b/test/sql/query.py
index 275bbe78c66e33ed01cb09c79826f716410fac26..bf178ae8f541e02553279a60f3327dfc9d4e0faf 100644 (file)
--- a/test/sql/query.py
+++ b/test/sql/query.py
@@ -664,7 +664,72 @@ class QueryTest(TestBase):
         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):

diff --git a/test/sql/select.py b/test/sql/select.py
index ea9f27cdf2dd06dcf124f774619e228d9b4b953b..aff4a8670e0ad73335e8b043773fc0857b3a72fd 100644 (file)
--- a/test/sql/select.py
+++ b/test/sql/select.py
@@ -836,16 +836,16 @@ FROM mytable, myothertable WHERE foo.id = foofoo(lala) AND datetime(foo) = Today
                             "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"'''
         )