I'll look into what happens there but you probably need to use the same
bindparam() object for now:
b = bindparam('username', type_=String)
tt.name == b
tt.username == b
On Mar 7, 2014, at 5:47 AM, Jas Per <[email protected]> wrote:
> I get an exception, when I try to use a bindparam variable twice in a query
> in MySql like this:
>
> TestTable.name == bindparam('username', type_ = String),
> TestTable.username == bindparam('username', type_ = String)
>
> - ubuntu 13.10 64bit
> - python 3.3.2
> - sqlalchemy 0.9.3
> - mysql 5.5.35
> - mysql-connector-python 1.1.4
>
> tested with sqlite and postgres as well, both do not fail. looks like mysql
> needs two separate parameters {'username_1': 'test','username_2': 'test'} ..?
> full test case:
>
> from sqlalchemy.engine import create_engine
> from sqlalchemy.sql.expression import select, and_, bindparam, insert
> from sqlalchemy.types import String,Integer
> from sqlalchemy.ext.declarative.api import declarative_base
> from sqlalchemy.sql.schema import Column
>
> def testCase(dbType):
> engine = connectDB(dbType)
> connection = engine.connect()
>
> Base = declarative_base()
>
> class TestTable(Base):
> __tablename__ = 'TestTable'
>
> ID = Column(Integer, primary_key=True)
> name = Column(String(255), nullable=False)
> username = Column(String(255), nullable=False)
>
> Base.metadata.create_all(bind=engine)
>
> insData = [{'name':'test','username':'test'}]
> connection.execute(insert(TestTable,insData))
>
> statement = select([TestTable]).where(and_(
> TestTable.name == bindparam('username', type_ = String),
> TestTable.username == bindparam('username', type_ =
> String)
> ))
>
> dbres =
> connection.execute(statement,username='test',usName='test').fetchall()
>
> assert len(dbres)
> print('PASSED: '+dbType)
> engine.dispose()
> connection.close()
>
> def connectDB(dbType):
> if dbType == 'sqlite':
> connectstring = 'sqlite://'
> engine = create_engine(connectstring, echo=True)
> elif dbType == 'postgres':
> connectstring = 'postgresql://postgres:test@localhost/'
> engine = create_engine(connectstring, echo=False)
> con = engine.connect()
> checkExists = con.execute("SELECT datname FROM pg_catalog.pg_database
> WHERE datname = 'testcasedb';").fetchall()
> if not len(checkExists):
> con.execute("commit")
> con.execute("CREATE DATABASE testcasedb")
> con.execute("commit")
> con.close()
> engine.dispose()
> engine = create_engine(connectstring+'testcasedb', echo=True)
> elif dbType == 'mysql':
> connectstring = 'mysql+mysqlconnector://root:test@localhost/'
> engine = create_engine(connectstring, echo=False)
> con = engine.connect()
> checkExists = con.execute("SELECT SCHEMA_NAME FROM
> INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME = 'testcasedb';").fetchall()
> if not len(checkExists):
> con.execute("CREATE DATABASE IF NOT EXISTS testcasedb;")#IF NOT
> EXISTS gives a warning on mysql that throws a DatabaseError in sqla
> con.close()
> engine.dispose()
> engine = create_engine(connectstring+'testcasedb', echo=True)
> return engine
>
>
> testCase('sqlite') #works
> testCase('postgres') #works
> testCase('mysql') #fails!
>
> --
> You received this message because you are subscribed to the Google Groups
> "sqlalchemy" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to [email protected].
> To post to this group, send email to [email protected].
> Visit this group at http://groups.google.com/group/sqlalchemy.
> For more options, visit https://groups.google.com/d/optout.
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
