diff options
| -rw-r--r-- | lib/sqlalchemy/ansisql.py | 32 | ||||
| -rw-r--r-- | lib/sqlalchemy/databases/mysql.py | 3 | ||||
| -rw-r--r-- | test/sql/quote.py | 50 |
3 files changed, 85 insertions, 0 deletions
diff --git a/lib/sqlalchemy/ansisql.py b/lib/sqlalchemy/ansisql.py index 3960482ee..94d69a8ce 100644 --- a/lib/sqlalchemy/ansisql.py +++ b/lib/sqlalchemy/ansisql.py @@ -974,6 +974,15 @@ class ANSIIdentifierPreparer(object): return value.replace('"', '""') + def _unescape_identifier(self, value): + """Canonicalize an escaped identifier. + + Subclasses should override this to provide database-dependent + unescaping behavior that reverses _escape_identifier. + """ + + return value.replace('""', '"') + def quote_identifier(self, value): """Quote an identifier. @@ -1097,4 +1106,27 @@ class ANSIIdentifierPreparer(object): else: return (self.format_table(table, use_schema=False), ) + def unformat_identifiers(self, identifiers): + """Unpack 'schema.table.column'-like strings into components.""" + + try: + r = self._r_identifiers + except AttributeError: + initial, final, escaped_final = \ + [re.escape(s) for s in + (self.initial_quote, self.final_quote, + self._escape_identifier(self.final_quote))] + r = re.compile( + r'(?:' + r'(?:%(initial)s((?:%(escaped)s|[^%(final)s])+)%(final)s' + r'|([^\.]+))(?=\.|$))+' % + { 'initial': initial, + 'final': final, + 'escaped': escaped_final }) + self._r_identifiers = r + + return [self._unescape_identifier(i) + for i in [a or b for a, b in r.findall(identifiers)]] + + dialect = ANSIDialect diff --git a/lib/sqlalchemy/databases/mysql.py b/lib/sqlalchemy/databases/mysql.py index 766748c62..0b9ab3531 100644 --- a/lib/sqlalchemy/databases/mysql.py +++ b/lib/sqlalchemy/databases/mysql.py @@ -1799,6 +1799,9 @@ class MySQLIdentifierPreparer(ansisql.ANSIIdentifierPreparer): def _escape_identifier(self, value): return value.replace('`', '``') + def _unescape_identifier(self, value): + return value.replace('``', '`') + def _fold_identifier_case(self, value): # TODO: determine MySQL's case folding rules # diff --git a/test/sql/quote.py b/test/sql/quote.py index 2fdf9dba0..eb0239124 100644 --- a/test/sql/quote.py +++ b/test/sql/quote.py @@ -137,6 +137,56 @@ class QuoteTest(PersistTest): x = lc_table1.select(distinct=True).alias("lala").select().scalar() finally: meta.drop_all() + +class PreparerTest(PersistTest): + """Test the db-agnostic quoting services of ANSIIdentifierPreparer.""" + + def test_unformat(self): + prep = ansisql.ANSIIdentifierPreparer(None) + unformat = prep.unformat_identifiers + + def a_eq(have, want): + if have != want: + print "Wanted %s" % want + print "Received %s" % have + self.assert_(have == want) + + a_eq(unformat('foo'), ['foo']) + a_eq(unformat('"foo"'), ['foo']) + a_eq(unformat("'foo'"), ["'foo'"]) + a_eq(unformat('foo.bar'), ['foo', 'bar']) + a_eq(unformat('"foo"."bar"'), ['foo', 'bar']) + a_eq(unformat('foo."bar"'), ['foo', 'bar']) + a_eq(unformat('"foo".bar'), ['foo', 'bar']) + a_eq(unformat('"foo"."b""a""r"."baz"'), ['foo', 'b"a"r', 'baz']) + + def test_unformat_custom(self): + class Custom(ansisql.ANSIIdentifierPreparer): + def __init__(self, dialect): + super(Custom, self).__init__(dialect, initial_quote='`', + final_quote='`') + def _escape_identifier(self, value): + return value.replace('`', '``') + def _unescape_identifier(self, value): + return value.replace('``', '`') + + prep = Custom(None) + unformat = prep.unformat_identifiers + + def a_eq(have, want): + if have != want: + print "Wanted %s" % want + print "Received %s" % have + self.assert_(have == want) + + a_eq(unformat('foo'), ['foo']) + a_eq(unformat('`foo`'), ['foo']) + a_eq(unformat(`'foo'`), ["'foo'"]) + a_eq(unformat('foo.bar'), ['foo', 'bar']) + a_eq(unformat('`foo`.`bar`'), ['foo', 'bar']) + a_eq(unformat('foo.`bar`'), ['foo', 'bar']) + a_eq(unformat('`foo`.bar'), ['foo', 'bar']) + a_eq(unformat('`foo`.`b``a``r`.`baz`'), ['foo', 'b`a`r', 'baz']) if __name__ == "__main__": testbase.main() |