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.
