本文整理汇总了Python中springpython.database.core.DatabaseTemplate类的典型用法代码示例。如果您正苦于以下问题：Python DatabaseTemplate类的具体用法？Python DatabaseTemplate怎么用？Python DatabaseTemplate使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了DatabaseTemplate类的16个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: createTables
def createTables(self):
self.createdTables = True
try:
self.factory = factory.MySQLConnectionFactory("springpython", "springpython", "localhost", "springpython")
dt = DatabaseTemplate(self.factory)
dt.execute("DROP TABLE IF EXISTS animal")
dt.execute("""
CREATE TABLE animal (
id serial PRIMARY KEY,
name VARCHAR(11),
category VARCHAR(20),
population SMALLINT
) ENGINE=innodb
""")
self.factory.commit()
except Exception, e:
print("""
!!! Can't run MySQLDatabaseTemplateTestCase !!!
This assumes you have executed some step like:
% sudo apt-get install mysql (Ubuntu)
% apt-get install mysql (Debian)
And then created a database for the spring python user:
% mysql -uroot
mysql> DROP DATABASE IF EXISTS springpython;
mysql> CREATE DATABASE springpython;
mysql> GRANT ALL ON springpython.* TO [email protected] IDENTIFIED BY 'springpython';
That should setup the springpython user to be able to create tables as needed for these test cases.
""")
raise e
开发者ID:alanfranz，项目名称:spring-python，代码行数:33，代码来源:databaseCoreTestCases.py
示例2: load_user
def load_user(self, username):
dt = DatabaseTemplate(self.dataSource)
users = dt.query(self.users_by_username_query, (username,), self.UsersByUsernameMapping())
if len(users) == 0:
raise UsernameNotFoundException("User not found")
user = users[0] # First item in list, first column of tuple, containing no GrantedAuthority[]
dbAuths = dt.query(
self.auth_by_username_query, (user.username,), self.AuthoritiesByUsernameMapping(self.role_prefix)
)
self.add_custom_authorities(user.username, dbAuths)
if len(dbAuths) == 0:
raise UsernameNotFoundException("User has no GrantedAuthority")
auths = [dbAuth for dbAuth in dbAuths]
return_username = user.username
if not self.username_based_pk:
return_username = username
self.logger.debug("Just fetched %s from the database" % user)
return User(return_username, user.password, user.enabled, True, True, True, auths)
开发者ID:ktan2020，项目名称:legacy-automation，代码行数:25，代码来源:dao.py
示例3: TransactionalBankWithNoTransactionalArguments
class TransactionalBankWithNoTransactionalArguments(object):
"""This sample application can be used to demonstrate the value of atomic operations. The transfer operation
must be wrapped in a transaction in order to perform correctly. Otherwise, any errors in the deposit will
allow the from-account to leak assets."""
def __init__(self, factory):
self.logger = logging.getLogger("springpythontest.testSupportClasses.TransactionalBankWithNoTransactionalArguments")
self.dt = DatabaseTemplate(factory)
def open(self, account_num):
self.logger.debug("Opening account %s with $0 balance." % account_num)
self.dt.execute("INSERT INTO account (account_num, balance) VALUES (?,?)", (account_num, 0))
def deposit(self, amount, account_num):
self.logger.debug("Depositing $%s into %s" % (amount, account_num))
rows = self.dt.execute("UPDATE account SET balance = balance + ? WHERE account_num = ?", (amount, account_num))
if rows == 0:
raise BankException("Account %s does NOT exist" % account_num)
def withdraw(self, amount, account_num):
self.logger.debug("Withdrawing $%s from %s" % (amount, account_num))
rows = self.dt.execute("UPDATE account SET balance = balance - ? WHERE account_num = ?", (amount, account_num))
if rows == 0:
raise BankException("Account %s does NOT exist" % account_num)
return amount
def balance(self, account_num):
return self.dt.query_for_object("SELECT balance FROM account WHERE account_num = ?", (account_num,), types.FloatType)
@transactional
def transfer(self, amount, from_account, to_account):
self.logger.debug("Transferring $%s from %s to %s." % (amount, from_account, to_account))
self.withdraw(amount, from_account)
self.deposit(amount, to_account)
开发者ID:alanfranz，项目名称:spring-python，代码行数:33，代码来源:testSupportClasses.py
示例4: testQueryingOracleWithValidlyFormattedArguments
def testQueryingOracleWithValidlyFormattedArguments(self):
cursor = self.mock()
cursor.expects(once()).method("execute")
cursor.expects(once()).method("fetchall").will(return_value([("workDir", "reloadDir", "archiveDir", "oid1")]))
conn = self.mock()
conn.expects(once()).method("cursor").will(return_value(cursor))
sys.modules["cx_Oracle"] = self.mock()
sys.modules["cx_Oracle"].expects(once()).method("connect").will(return_value(conn))
connection_factory = factory.cxoraConnectionFactory(username="foo", password="bar", hostname="localhost", db="mock")
dt = DatabaseTemplate(connection_factory)
dt.query("""
SELECT
impcarrcfg.paystat_work_dir,
impcarrcfg.paystat_reload_dir,
impcarrcfg.paystat_archive_dir,
impcarrcfg.oid
FROM impcarrcfg, carr, lklabelsys
WHERE (lklabelsys.oid = impcarrcfg.lklabelsys_oid)
and (carr.oid = impcarrcfg.carr_oid )
and (carr.oid = :carr_oid and lklabelsys.oid = :lklabelsys_oid)
""",
{'carr_oid':5, 'lklabelsys_oid':5},
testSupportClasses.ImpFilePropsRowMapper())
del(sys.modules["cx_Oracle"])
开发者ID:alanfranz，项目名称:spring-python，代码行数:29，代码来源:databaseCoreTestCases.py
示例5: Bank
class Bank(object):
def __init__(self, connectionFactory):
self.factory = connectionFactory
self.dt = DatabaseTemplate(connectionFactory)
@transactional
def withdraw(self, amount, account):
DECREASE_STMT = """
UPDATE account SET balance = balance - ? WHERE account_name = ?
"""
balance = self.get_balance(account)
if (balance < amount):
raise Exception("not enough balance (only $ %s) in %s" % (balance, account) )
print "[DEBUG] before DECREASE_STMT"
rows = self.dt.execute(DECREASE_STMT, (amount, account))
print "[DEBUG] rows = %d" % rows
if (rows == 0):
raise Exception("no such account %s" % account)
@transactional
def deposit(self, amount, account):
print "[DEBUG] deposit() begins"
INCREASE_STMT = """
UPDATE account SET balance = balance + ? WHERE account_name = ?
"""
print "[DEBUG] before INCREASE_STMT"
rows = self.dt.execute(INCREASE_STMT, (amount, account))
print "[DEBUG] rows = %d" % rows
if (rows == 0):
raise Exception("no such account %s" % account)
@transactional
def transfer(self, amount, from_account, to_account):
if (not self.is_valid_account(from_account)):
raise Exception("invalid account: %s" % from_account)
if (not self.is_valid_account(to_account)):
raise Exception("invalid account: %s" % to_account)
# TODO: what if step 1 succeeds, but step 2 fails
self.withdraw(amount, from_account)
self.deposit(amount, to_account)
@transactional(["PROPAGATION_SUPPORTS"])
def get_balance(self, account_name):
GET_BALANCE_STMT = """
SELECT balance FROM account WHERE account_name = ?
"""
balance = self.dt.query_for_object(GET_BALANCE_STMT, (account_name,), types.LongType)
return balance
def is_valid_account(self, account):
account_list = ("saving_111", "checking_222")
return account in account_list
开发者ID:jeffshaw，项目名称:spring-python-example，代码行数:58，代码来源:Bank.py
示例6: createTables
def createTables(self):
self.createdTables = True
try:
try:
os.remove(self.db_filename)
except OSError, e:
pass
self.factory = factory.Sqlite3ConnectionFactory(self.db_filename)
dt = DatabaseTemplate(self.factory)
dt.execute("DROP TABLE IF EXISTS animal")
dt.execute("DROP TABLE IF EXISTS account")
dt.execute("""
CREATE TABLE animal (
id serial PRIMARY KEY,
name VARCHAR(11)
)
""")
dt.execute("""
CREATE TABLE account (
id serial PRIMARY KEY,
account_num VARCHAR(11),
balance FLOAT(10)
)
""")
self.factory.commit()
开发者ID:jeffbrown，项目名称:spring-python，代码行数:28，代码来源:databaseTransactionTestCases.py
示例7: testIoCGeneralQuery
def testIoCGeneralQuery(self):
appContext = ApplicationContext(XMLConfig("support/databaseTestApplicationContext.xml"))
mockConnectionFactory = appContext.get_object("mockConnectionFactory")
mockConnectionFactory.stubConnection.mockCursor = self.mock
self.mock.expects(once()).method("execute")
self.mock.expects(once()).method("fetchall").will(return_value([("me", "myphone")]))
databaseTemplate = DatabaseTemplate(connection_factory = mockConnectionFactory)
results = databaseTemplate.query("select * from foobar", rowhandler=testSupportClasses.SampleRowMapper())
开发者ID:alanfranz，项目名称:spring-python，代码行数:11，代码来源:databaseCoreTestCases.py
示例8: setupDatabase
def setupDatabase():
"""Figure out what type of database exists, and then set it up."""
connectionFactory = tryMySQL()
if connectionFactory is None:
raise Exception("+++ Could not setup MySQL. We don't support any others yet.")
databaseTemplate = DatabaseTemplate(connectionFactory)
for sqlStatement in [line.strip() for line in open("db/populateDB.txt").readlines()
if line.strip() != ""]:
databaseTemplate.execute(sqlStatement)
print "+++ Database is setup."
开发者ID:Cophy08，项目名称:spring-python，代码行数:14，代码来源:configure.py
示例9: setUp
def setUp(self):
if not self.createdTables:
self.createTables()
self.createTables()
self.dt = DatabaseTemplate(self.factory)
self.dt.execute("DELETE FROM animal")
self.dt.execute("DELETE FROM account")
self.factory.commit()
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 0)
self.transactionManager = ConnectionFactoryTransactionManager(self.factory)
self.transactionTemplate = TransactionTemplate(self.transactionManager)
开发者ID:jeffbrown，项目名称:spring-python，代码行数:11，代码来源:databaseTransactionTestCases.py
示例10: setUp
def setUp(self):
if not self.createdTables:
self.createTables()
self.databaseTemplate = DatabaseTemplate(self.factory)
self.databaseTemplate.execute("DELETE FROM animal")
self.factory.commit()
self.assertEquals(len(self.databaseTemplate.query_for_list("SELECT * FROM animal")), 0)
self.databaseTemplate.execute("INSERT INTO animal (name, category, population) VALUES ('snake', 'reptile', 1)")
self.databaseTemplate.execute("INSERT INTO animal (name, category, population) VALUES ('racoon', 'mammal', 0)")
self.databaseTemplate.execute ("INSERT INTO animal (name, category, population) VALUES ('black mamba', 'kill_bill_viper', 1)")
self.databaseTemplate.execute ("INSERT INTO animal (name, category, population) VALUES ('cottonmouth', 'kill_bill_viper', 1)")
self.factory.commit()
self.assertEquals(len(self.databaseTemplate.query_for_list("SELECT * FROM animal")), 4)
开发者ID:alanfranz，项目名称:spring-python，代码行数:13，代码来源:databaseCoreTestCases.py
示例11: AbstractTransactionTestCase
class AbstractTransactionTestCase(unittest.TestCase):
def __init__(self, methodName='runTest'):
unittest.TestCase.__init__(self, methodName)
self.factory = None
self.createdTables = False
def setUp(self):
if not self.createdTables:
self.createTables()
self.createTables()
self.dt = DatabaseTemplate(self.factory)
self.dt.execute("DELETE FROM animal")
self.dt.execute("DELETE FROM account")
self.factory.commit()
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 0)
self.transactionManager = ConnectionFactoryTransactionManager(self.factory)
self.transactionTemplate = TransactionTemplate(self.transactionManager)
def tearDown(self):
self.factory.getConnection().rollback()
def testInsertingRowsIntoTheDatabase(self):
rows = self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('black mamba',))
self.assertEquals(rows, 1)
name = self.dt.query_for_object("SELECT name FROM animal WHERE name = 'black mamba'", required_type=types.StringType)
self.assertEquals(name, "black mamba")
def testInsertingTwoRowsWithoutaTransactionButManuallyCommitted(self):
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('black mamba',))
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('copperhead',))
self.factory.commit()
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 2)
def testInsertingTwoRowsWithoutaTransactionButManuallyRolledBack(self):
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('black mamba',))
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('copperhead',))
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 2)
self.dt.connection_factory.getConnection().rollback()
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 0)
def testInsertingTwoRowsWithaTransactionAndNoErrorsAndNoResults(self):
class txDefinition(TransactionCallbackWithoutResult):
def do_in_tx_without_result(s, status):
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('black mamba',))
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('copperhead',))
self.transactionTemplate.execute(txDefinition())
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 2)
def testInsertingTwoRowsWithaTransactionAndAnIntermediateErrorAndNoResults(self):
class txDefinition(TransactionCallbackWithoutResult):
def do_in_tx_without_result(s, status):
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('black mamba',))
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 1)
raise DataAccessException("This should break the transaction, and rollback the insert.")
self.assertRaises(DataAccessException, self.transactionTemplate.execute, txDefinition())
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 0)
def testInsertingTwoRowsWithaTransactionAndNoErrorsAndResults(self):
class txDefinition(TransactionCallback):
def do_in_transaction(s, status):
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('black mamba',))
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('copperhead',))
results = self.dt.query_for_object("SELECT name FROM animal WHERE name like 'c%'", required_type=types.StringType)
return results
self.assertEquals(self.transactionTemplate.execute(txDefinition()), "copperhead")
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 2)
def testInsertingTwoRowsWithaTransactionAndAnIntermediateErrorAndResults(self):
class txDefinition(TransactionCallback):
def do_in_transaction(s, status):
self.dt.execute("INSERT INTO animal (name) VALUES (?)", ('black mamba'))
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 1)
raise DataAccessException("This should break the transaction, and rollback the insert.")
self.assertRaises(DataAccessException, self.transactionTemplate.execute, txDefinition())
self.assertEquals(len(self.dt.query_for_list("SELECT * FROM animal")), 0)
def testDeclarativeTransactions(self):
appContext = ApplicationContext(DatabaseTxTestAppContext(self.factory))
bank = appContext.get_object("bank")
bank.open("Checking")
bank.open("Savings")
bank.deposit(125.00, "Checking")
self.assertEquals(bank.balance("Checking"), 125.00)
bank.deposit(250.00, "Savings")
self.assertEquals(bank.balance("Savings"), 250.00)
bank.transfer(25.00, "Savings", "Checking")
self.assertEquals(bank.balance("Savings"), 225.00)
self.assertEquals(bank.balance("Checking"), 150.00)
bank.withdraw(10.00, "Checking")
#.........这里部分代码省略.........
开发者ID:jeffbrown，项目名称:spring-python，代码行数:101，代码来源:databaseTransactionTestCases.py
示例12: __init__
def __init__(self, connectionFactory):
self.factory = connectionFactory
self.dt = DatabaseTemplate(connectionFactory)
开发者ID:jeffshaw，项目名称:spring-python-example，代码行数:3，代码来源:Bank.py
示例13: testIoCGeneralQueryWithDictionaryRowMapper
def testIoCGeneralQueryWithDictionaryRowMapper(self):
appContext = ApplicationContext(XMLConfig("support/databaseTestSqliteApplicationContext.xml"))
factory = appContext.get_object("connection_factory")
databaseTemplate = DatabaseTemplate(factory)
databaseTemplate.execute("DROP TABLE IF EXISTS animal")
databaseTemplate.execute("""
CREATE TABLE animal (
id serial PRIMARY KEY,
name VARCHAR(11),
category VARCHAR(20),
population integer
)
""")
factory.commit()
databaseTemplate.execute("DELETE FROM animal")
factory.commit()
self.assertEquals(len(databaseTemplate.query_for_list("SELECT * FROM animal")), 0)
databaseTemplate.execute("INSERT INTO animal (name, category, population) VALUES ('snake', 'reptile', 1)")
databaseTemplate.execute("INSERT INTO animal (name, category, population) VALUES ('racoon', 'mammal', 0)")
databaseTemplate.execute ("INSERT INTO animal (name, category, population) VALUES ('black mamba', 'kill_bill_viper', 1)")
databaseTemplate.execute ("INSERT INTO animal (name, category, population) VALUES ('cottonmouth', 'kill_bill_viper', 1)")
factory.commit()
self.assertEquals(len(databaseTemplate.query_for_list("SELECT * FROM animal")), 4)
results = databaseTemplate.query("select * from animal", rowhandler=DictionaryRowMapper())
开发者ID:alanfranz，项目名称:spring-python，代码行数:27，代码来源:databaseCoreTestCases.py
示例14: __init__
def __init__(self, factory):
self.logger = logging.getLogger("springpythontest.testSupportClasses.Bank")
self.dt = DatabaseTemplate(factory)
开发者ID:alanfranz，项目名称:spring-python，代码行数:3，代码来源:testSupportClasses.py
示例15: DatabaseTemplateMockTestCase
class DatabaseTemplateMockTestCase(MockTestCase):
"""Testing the DatabaseTemplate utilizes stubbing and mocking, in order to isolate from different
vendor implementations. This reduces the overhead in making changes to core functionality."""
def setUp(self):
self.mock = self.mock()
connection_factory = testSupportClasses.StubDBFactory()
connection_factory.stubConnection.mockCursor = self.mock
self.databaseTemplate = DatabaseTemplate(connection_factory)
def testProgrammaticallyInstantiatingAnAbstractDatabaseTemplate(self):
emptyTemplate = DatabaseTemplate()
self.assertRaises(AttributeError, emptyTemplate.query, "sql query shouldn't work", None)
def testProgrammaticHandlingInvalidRowHandler(self):
self.mock.expects(once()).method("execute")
self.mock.expects(once()).method("fetchall").will(return_value([("me", "myphone")]))
self.assertRaises(AttributeError, self.databaseTemplate.query, "select * from foobar", rowhandler=testSupportClasses.InvalidCallbackHandler())
def testProgrammaticHandlingImproperRowHandler(self):
self.mock.expects(once()).method("execute")
self.mock.expects(once()).method("fetchall").will(return_value([("me", "myphone")]))
self.assertRaises(TypeError, self.databaseTemplate.query, "select * from foobar", rowhandler=testSupportClasses.ImproperCallbackHandler())
def testProgrammaticHandlingValidDuckTypedRowHandler(self):
self.mock.expects(once()).method("execute")
self.mock.expects(once()).method("fetchall").will(return_value([("me", "myphone")]))
results = self.databaseTemplate.query("select * from foobar", rowhandler=testSupportClasses.ValidHandler())
def testIoCGeneralQuery(self):
appContext = ApplicationContext(XMLConfig("support/databaseTestApplicationContext.xml"))
mockConnectionFactory = appContext.get_object("mockConnectionFactory")
mockConnectionFactory.stubConnection.mockCursor = self.mock
self.mock.expects(once()).method("execute")
self.mock.expects(once()).method("fetchall").will(return_value([("me", "myphone")]))
databaseTemplate = DatabaseTemplate(connection_factory = mockConnectionFactory)
results = databaseTemplate.query("select * from foobar", rowhandler=testSupportClasses.SampleRowMapper())
def testProgrammaticStaticQuery(self):
self.assertRaises(ArgumentMustBeNamed, self.databaseTemplate.query, "select * from animal", testSupportClasses.AnimalRowMapper())
self.mock.expects(once()).method("execute").id("#1")
self.mock.expects(once()).method("fetchall").will(return_value([('snake', 'reptile', 1), ('racoon', 'mammal', 1)])).id("#2").after("#1")
animals = self.databaseTemplate.query("select * from animal", rowhandler=testSupportClasses.AnimalRowMapper())
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
self.assertEquals(animals[1].name, "racoon")
self.assertEquals(animals[1].category, "mammal")
def testProgrammaticQueryWithBoundArguments(self):
self.mock.expects(once()).method("execute").id("#1")
self.mock.expects(once()).method("fetchall").will(return_value([('snake', 'reptile', 1)])).id("#2").after("#1")
self.mock.expects(once()).method("execute").id("#3").after("#2")
self.mock.expects(once()).method("fetchall").will(return_value([('snake', 'reptile', 1)])).id("#4").after("#3")
animals = self.databaseTemplate.query("select * from animal where name = %s", ("snake",), testSupportClasses.AnimalRowMapper())
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
animals = self.databaseTemplate.query("select * from animal where name = ?", ("snake",), testSupportClasses.AnimalRowMapper())
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
def testProgrammaticStaticQueryForList(self):
self.mock.expects(once()).method("execute").id("#1")
self.mock.expects(once()).method("fetchall").will(return_value([('snake', 'reptile', 1), ('racoon', 'mammal', 1)])).id("#2").after("#1")
animals = self.databaseTemplate.query_for_list("select * from animal")
self.assertEquals(animals[0][0], "snake")
self.assertEquals(animals[0][1], "reptile")
self.assertEquals(animals[1][0], "racoon")
self.assertEquals(animals[1][1], "mammal")
def testProgrammaticQueryForListWithBoundArguments(self):
self.mock.expects(once()).method("execute").id("#1")
self.mock.expects(once()).method("fetchall").will(return_value([('snake', 'reptile', 1)])).id("#2").after("#1")
self.mock.expects(once()).method("execute").id("#3").after("#2")
self.mock.expects(once()).method("fetchall").will(return_value([('snake', 'reptile', 1)])).id("#4").after("#3")
animals = self.databaseTemplate.query_for_list("select * from animal where name = %s", ("snake",))
self.assertEquals(animals[0][0], "snake")
self.assertEquals(animals[0][1], "reptile")
animals = self.databaseTemplate.query_for_list("select * from animal where name = ?", ("snake",))
self.assertEquals(animals[0][0], "snake")
self.assertEquals(animals[0][1], "reptile")
def testProgrammaticQueryForListWithBoundArgumentsNotProperlyTuplized(self):
self.assertRaises(InvalidArgumentType, self.databaseTemplate.query_for_list, "select * from animal where name = %s", "snake")
self.assertRaises(InvalidArgumentType, self.databaseTemplate.query_for_list, "select * from animal where name = ?", "snake")
def testProgrammaticStaticQueryForInt(self):
self.mock.expects(once()).method("execute").id("#1")
#.........这里部分代码省略.........
开发者ID:alanfranz，项目名称:spring-python，代码行数:101，代码来源:databaseCoreTestCases.py
示例16: AbstractDatabaseTemplateTestCase
class AbstractDatabaseTemplateTestCase(unittest.TestCase):
def __init__(self, methodName='runTest'):
unittest.TestCase.__init__(self, methodName)
self.factory = None
self.createdTables = False
def setUp(self):
if not self.createdTables:
self.createTables()
self.databaseTemplate = DatabaseTemplate(self.factory)
self.databaseTemplate.execute("DELETE FROM animal")
self.factory.commit()
self.assertEquals(len(self.databaseTemplate.query_for_list("SELECT * FROM animal")), 0)
self.databaseTemplate.execute("INSERT INTO animal (name, category, population) VALUES ('snake', 'reptile', 1)")
self.databaseTemplate.execute("INSERT INTO animal (name, category, population) VALUES ('racoon', 'mammal', 0)")
self.databaseTemplate.execute ("INSERT INTO animal (name, category, population) VALUES ('black mamba', 'kill_bill_viper', 1)")
self.databaseTemplate.execute ("INSERT INTO animal (name, category, population) VALUES ('cottonmouth', 'kill_bill_viper', 1)")
self.factory.commit()
self.assertEquals(len(self.databaseTemplate.query_for_list("SELECT * FROM animal")), 4)
def tearDown(self):
self.factory.rollback()
def testProgrammaticallyInstantiatingAnAbstractDatabaseTemplate(self):
emptyTemplate = DatabaseTemplate()
self.assertRaises(AttributeError, emptyTemplate.query, "sql query shouldn't work", None)
def testProgrammaticHandlingInvalidRowHandler(self):
self.assertRaises(AttributeError, self.databaseTemplate.query, "select * from animal", rowhandler=testSupportClasses.InvalidCallbackHandler())
def testProgrammaticHandlingImproperRowHandler(self):
self.assertRaises(TypeError, self.databaseTemplate.query, "select * from animal", rowhandler=testSupportClasses.ImproperCallbackHandler())
def testProgrammaticHandlingValidDuckTypedRowHandler(self):
results = self.databaseTemplate.query("select * from animal", rowhandler=testSupportClasses.ValidHandler())
def testProgrammaticStaticQuery(self):
self.assertRaises(ArgumentMustBeNamed, self.databaseTemplate.query, "select * from animal", testSupportClasses.AnimalRowMapper())
animals = self.databaseTemplate.query("select name, category from animal", rowhandler=testSupportClasses.AnimalRowMapper())
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
self.assertEquals(animals[1].name, "racoon")
self.assertEquals(animals[1].category, "mammal")
def testProgrammaticStaticQueryWithSimpleRowMapper(self):
animals = self.databaseTemplate.query("select name, category from animal", rowhandler=SimpleRowMapper(testSupportClasses.Animal))
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
self.assertEquals(animals[1].name, "racoon")
self.assertEquals(animals[1].category, "mammal")
def testProgrammaticStaticQueryWithDictionaryRowMapper(self):
animals = self.databaseTemplate.query("select name, category from animal", rowhandler=DictionaryRowMapper())
self.assertEquals(animals[0]["name"], "snake")
self.assertEquals(animals[0]["category"], "reptile")
self.assertEquals(animals[1]["name"], "racoon")
self.assertEquals(animals[1]["category"], "mammal")
def testProgrammaticQueryWithBoundArguments(self):
animals = self.databaseTemplate.query("select name, category from animal where name = %s", ("snake",), testSupportClasses.AnimalRowMapper())
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
animals = self.databaseTemplate.query("select name, category from animal where name = ?", ("snake",), testSupportClasses.AnimalRowMapper())
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
def testProgrammaticQueryWithBoundArgumentsWithSimpleRowMapper(self):
animals = self.databaseTemplate.query("select name, category from animal where name = %s", ("snake",), SimpleRowMapper(testSupportClasses.Animal))
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
animals = self.databaseTemplate.query("select name, category from animal where name = ?", ("snake",), SimpleRowMapper(testSupportClasses.Animal))
self.assertEquals(animals[0].name, "snake")
self.assertEquals(animals[0].category, "reptile")
def testProgrammaticQueryWithBoundArgumentsWithDictionaryRowMapper(self):
animals = self.databaseTemplate.query("select name, category from animal where name = %s", ("snake",), DictionaryRowMapper())
self.assertEquals(animals[0]["name"], "snake")
self.assertEquals(animals[0]["category"], "reptile")
animals = self.databaseTemplate.query("select name, category from animal where name = ?", ("snake",), DictionaryRowMapper())
self.assertEquals(animals[0]["name"], "snake")
self.assertEquals(animals[0]["category"], "reptile")
def testProgrammaticStaticQueryForList(self):
animals = self.databaseTemplate.query_for_list("select name, category from animal")
self.assertEquals(animals[0][0], "snake")
self.assertEquals(animals[0][1], "reptile")
self.assertEquals(animals[1][0], "racoon")
self.assertEquals(animals[1][1], "mammal")
def testProgrammaticQueryForListWithBoundArguments(self):
animals = self.databaseTemplate.query_for_list("select name, category from animal where name = %s", ("snake",))
self.assertEquals(animals[0][0], "snake")
self.assertEquals(animals[0][1], "reptile")
animals = self.databaseTemplate.query_for_list("select name, category from animal where name = ?", ("snake",))
self.assertEquals(animals[0][0], "snake")
#.........这里部分代码省略.........
开发者ID:alanfranz，项目名称:spring-python，代码行数:101，代码来源:databaseCoreTestCases.py
注：本文中的springpython.database.core.DatabaseTemplate类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论