From 4f8a560d6f02c60ac85e73ad90d2ffb4a183d9f0 Mon Sep 17 00:00:00 2001
From: Jason Kirtland <jek@discorporate.us>
Date: Thu, 9 Aug 2007 22:31:44 +0000
Subject: [PATCH] Fixed detection of internal '$' characters in :bind$params
 (Applied patch in [ticket:719], thanks!)

---
 CHANGES                   |  1 +
 lib/sqlalchemy/ansisql.py |  2 +-
 test/sql/query.py         | 20 +++++++++++++++++++-
 3 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/CHANGES b/CHANGES
index 621ea0db09..1d952c8d70 100644
--- 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 9994d52889..e7146fb3d9 100644
--- 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 8af5aafeab..6880b33f2c 100644
--- 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')
-- 
2.47.2