Fixed detection of internal '$' characters in :bind$params (Applied patch in [ticket...

diff --git a/CHANGES b/CHANGES
index 621ea0db09fed4680114e31938a7336db999b934..1d952c8d708737d592f3e7087d3f707cb1945ab3 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -2,6 +2,7 @@
 - sql
     - tweak DISTINCT precedence for clauses like
       `func.count(t.c.col.distinct())`
+    - Fixed detection of internal '$' characters in :bind$params [ticket:719]
 - orm
     - added a check for joining from A->B using join(), along two
       different m2m tables.  this raises an error in 0.3 but is 

diff --git a/lib/sqlalchemy/ansisql.py b/lib/sqlalchemy/ansisql.py
index 9994d528895e9db1f8fe8524ebb3a979ee3ccd7a..e7146fb3d9fd05063781f3901ee844d779b1f9d2 100644 (file)
--- a/lib/sqlalchemy/ansisql.py
+++ b/lib/sqlalchemy/ansisql.py
@@ -170,7 +170,7 @@ class ANSICompiler(sql.Compiled):
         # this re will search for params like :param
         # it has a negative lookbehind for an extra ':' so that it doesnt match
         # postgres '::text' tokens
-        match = re.compile(r'(?<!:):([\w_]+)', re.UNICODE)
+        match = re.compile(r'(?<!:):([\w_\$]+)', re.UNICODE)
         if self.paramstyle=='pyformat':
             self.strings[self.statement] = match.sub(lambda m:'%(' + m.group(1) +')s', self.strings[self.statement])
         elif self.positional:

diff --git a/test/sql/query.py b/test/sql/query.py
index 8af5aafeabafdf6862cf4c8539a0dc43bc58687e..6880b33f2c66e4b95adce47a7e0b4b4701d8c607 100644 (file)
--- a/test/sql/query.py
+++ b/test/sql/query.py
@@ -7,7 +7,7 @@ import sqlalchemy.databases.sqlite as sqllite
 import tables
 from sqlalchemy import *
 from sqlalchemy.engine import ResultProxy, RowProxy
-from sqlalchemy import exceptions
+from sqlalchemy import exceptions, ansisql
 
 class QueryTest(PersistTest):
     
@@ -183,6 +183,24 @@ class QueryTest(PersistTest):
         s = self.users.select(self.users.c.user_name==u)
         r = s.execute(someshortname='fred').fetchall()
         assert len(r) == 1
+
+    def test_bindparam_detection(self):
+        dialect = ansisql.ANSIDialect(default_paramstyle='qmark')
+        prep = lambda q: str(dialect.compile(sql.text(q)))
+
+        def a_eq(got, wanted):
+            if got != wanted:
+                print "Wanted %s" % wanted
+                print "Received %s" % got
+            self.assert_(got == wanted)
+
+        a_eq(prep('select foo'), 'select foo')
+        a_eq(prep(":this :that"), "? ?")
+        a_eq(prep("(:this),(:that :other)"), "(?),(? ?)")
+        a_eq(prep("(:that_ :other)"), "(? ?)")
+        a_eq(prep("(:that_other)"), "(?)")
+        a_eq(prep("(:that$other)"), "(?)")
+        a_eq(prep(".:that$ :other."), ".? ?.")
         
     def testdelete(self):
         self.users.insert().execute(user_id = 7, user_name = 'jack')