From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Sat, 30 Nov 2013 00:06:33 +0000 (-0500)
Subject: - add support for bindparam() called from AsFromText
X-Git-Tag: rel_0_9_0~76
X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=207fafe7e35b945a41e139b5507a8d73c7b019db;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

- add support for bindparam() called from AsFromText
- get PG dialect to work around "no nonexistent binds" rule for now,
though we might want to reconsider this behavior
---

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 7ad266b58c..8bcfcbf7cf 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1545,12 +1545,6 @@ class PGDialect(default.DefaultDialect):
         return bool(cursor.first())
 
     def has_type(self, connection, type_name, schema=None):
-        bindparams = [
-            sql.bindparam('typname',
-                util.text_type(type_name), type_=sqltypes.Unicode),
-            sql.bindparam('nspname',
-                util.text_type(schema), type_=sqltypes.Unicode),
-            ]
         if schema is not None:
             query = """
             SELECT EXISTS (
@@ -1560,6 +1554,7 @@ class PGDialect(default.DefaultDialect):
                 AND n.nspname = :nspname
                 )
                 """
+            query = sql.text(query)
         else:
             query = """
             SELECT EXISTS (
@@ -1568,7 +1563,17 @@ class PGDialect(default.DefaultDialect):
                 AND pg_type_is_visible(t.oid)
                 )
                 """
-        cursor = connection.execute(sql.text(query, bindparams=bindparams))
+            query = sql.text(query)
+        query = query.bindparams(
+                sql.bindparam('typname',
+                    util.text_type(type_name), type_=sqltypes.Unicode),
+                )
+        if schema is not None:
+            query = query.bindparams(
+                    sql.bindparam('nspname',
+                        util.text_type(schema), type_=sqltypes.Unicode),
+                    )
+        cursor = connection.execute(query)
         return bool(cursor.scalar())
 
     def _get_server_version_info(self, connection):
@@ -1608,12 +1613,10 @@ class PGDialect(default.DefaultDialect):
         table_name = util.text_type(table_name)
         if schema is not None:
             schema = util.text_type(schema)
-        s = sql.text(query, bindparams=[
-            sql.bindparam('table_name', type_=sqltypes.Unicode),
-            sql.bindparam('schema', type_=sqltypes.Unicode)
-            ],
-            typemap={'oid': sqltypes.Integer}
-        )
+        s = sql.text(query).bindparams(table_name=sqltypes.Unicode)
+        s = s.columns(oid=sqltypes.Integer)
+        if schema:
+            s = s.bindparams(sql.bindparam('schema', type_=sqltypes.Unicode))
         c = connection.execute(s, table_name=table_name, schema=schema)
         table_oid = c.scalar()
         if table_oid is None:
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index b850e14655..79e341b778 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -2968,6 +2968,10 @@ class TextAsFrom(SelectBase):
     def _bind(self):
         return self.element._bind
 
+    @_generative
+    def bindparams(self, *binds, **bind_as_values):
+        self.element = self.element.bindparams(*binds, **bind_as_values)
+
     def _populate_column_collection(self):
         for c in self.column_args:
             c._make_proxy(self)
diff --git a/test/sql/test_text.py b/test/sql/test_text.py
index af9f8db050..827f4dfb98 100644
--- a/test/sql/test_text.py
+++ b/test/sql/test_text.py
@@ -345,4 +345,16 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
             s,
             "SELECT mytable.myid, (select id from user) AS anon_1 "
             "FROM mytable WHERE mytable.myid = (select id from user)"
+        )
+
+    def test_build_bindparams(self):
+        t = text("select id from user :foo :bar :bat")
+        t = t.bindparams(bindparam("foo", type_=Integer))
+        t = t.columns(id=Integer)
+        t = t.bindparams(bar=String)
+        t = t.bindparams(bindparam('bat', value='bat'))
+
+        eq_(
+            set(t.element._bindparams),
+            set(["bat", "foo", "bar"])
         )
\ No newline at end of file