diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2013-11-29 19:06:33 -0500 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2013-11-29 19:06:33 -0500 |
commit | 207fafe7e35b945a41e139b5507a8d73c7b019db (patch) | |
tree | 36a859de332c9d4a94b8333dd18ef29aeb610c06 | |
parent | 58fab1be984a42400e67bdf7f15acd5a6d7c4b3c (diff) | |
download | sqlalchemy-207fafe7e35b945a41e139b5507a8d73c7b019db.tar.gz |
- 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
-rw-r--r-- | lib/sqlalchemy/dialects/postgresql/base.py | 29 | ||||
-rw-r--r-- | lib/sqlalchemy/sql/selectable.py | 4 | ||||
-rw-r--r-- | test/sql/test_text.py | 12 |
3 files changed, 32 insertions, 13 deletions
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py index 7ad266b58..8bcfcbf7c 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 b850e1465..79e341b77 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 af9f8db05..827f4dfb9 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 |