diff options
| -rw-r--r-- | CHANGES | 6 | ||||
| -rw-r--r-- | lib/sqlalchemy/sql/compiler.py | 10 | ||||
| -rw-r--r-- | lib/sqlalchemy/sql/expression.py | 22 | ||||
| -rw-r--r-- | test/sql/query.py | 67 | ||||
| -rw-r--r-- | test/sql/select.py | 8 |
5 files changed, 96 insertions, 17 deletions
@@ -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 d5c85d71d..8622aeea4 100644 --- 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 f527c6351..48b1d0687 100644 --- 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 275bbe78c..bf178ae8f 100644 --- 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 ea9f27cdf..aff4a8670 100644 --- 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"''' ) |