- 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 7ad266b58c3e700a01a956c64b254780b590dd8b..8bcfcbf7cfe851d11dc7d9ab41f5f7ba674cf380 100644 (file)
--- 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 b850e146554103680ac5db7329995ca16148d895..79e341b778d75d0a81425b831c06df92ee1e6c7e 100644 (file)
--- 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 af9f8db050d3231d48651c0034d77d8915f5cf5b..827f4dfb982ba4cf10bb1c6f050fb6bd983be9ae 100644 (file)
--- 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