[Zope3-checkins] SVN: Zope3/trunk/src/zope/app/rdb/ Added custom
executemany() method for cursors created through GadflyAdapter.
Dmitry Vasiliev
dima at hlabs.spb.ru
Fri Sep 9 07:13:12 EDT 2005
Log message for revision 38419:
Added custom executemany() method for cursors created through GadflyAdapter.
See http://www.zope.org/Collectors/Zope3-dev/445 for discussion.
Changed:
U Zope3/trunk/src/zope/app/rdb/__init__.py
U Zope3/trunk/src/zope/app/rdb/gadflyda.py
U Zope3/trunk/src/zope/app/rdb/interfaces.py
U Zope3/trunk/src/zope/app/rdb/tests/test_gadflyadapter.py
-=-
Modified: Zope3/trunk/src/zope/app/rdb/__init__.py
===================================================================
--- Zope3/trunk/src/zope/app/rdb/__init__.py 2005-09-09 08:40:13 UTC (rev 38418)
+++ Zope3/trunk/src/zope/app/rdb/__init__.py 2005-09-09 11:13:11 UTC (rev 38419)
@@ -31,7 +31,7 @@
from zope.interface import implements
from zope.app.container.contained import Contained
-from zope.app.rdb.interfaces import DatabaseException
+from zope.app.rdb.interfaces import DatabaseException, DatabaseAdapterError
from zope.app.rdb.interfaces import IResultSet
from zope.app.rdb.interfaces import IZopeConnection, IZopeCursor
from zope.app.rdb.interfaces import IManageableZopeDatabaseAdapter
@@ -104,10 +104,6 @@
return cmp(len(self), len(other))
-class DatabaseAdapterError(Exception):
- pass
-
-
class ZopeDatabaseAdapter(Persistent, Contained):
implements(IManageableZopeDatabaseAdapter)
@@ -152,6 +148,8 @@
try:
self._v_connection = ZopeConnection(
self._connection_factory(), self)
+ except DatabaseException:
+ raise
# Note: I added the general Exception, since the DA can return
# implementation-specific errors. But we really want to catch all
# issues at this point, so that we can convert it to a
@@ -294,6 +292,8 @@
"""Executes an operation, registering the underlying
connection with the transaction system. """
operation, parameters = self._prepareOperation(operation, parameters)
+ # If executemany() is not defined pass parameters
+ # to execute() as defined by DB API v.1
method = getattr(self.cursor, "executemany", self.cursor.execute)
self.connection.registerForTxn()
return method(operation, parameters)
Modified: Zope3/trunk/src/zope/app/rdb/gadflyda.py
===================================================================
--- Zope3/trunk/src/zope/app/rdb/gadflyda.py 2005-09-09 08:40:13 UTC (rev 38418)
+++ Zope3/trunk/src/zope/app/rdb/gadflyda.py 2005-09-09 11:13:11 UTC (rev 38419)
@@ -19,17 +19,39 @@
import os
from zope.app.rdb import ZopeDatabaseAdapter, parseDSN
-from zope.app.rdb import DatabaseAdapterError
+from zope.app.rdb import DatabaseException, DatabaseAdapterError
+from zope.app.rdb import ZopeConnection, ZopeCursor
GadflyError = DatabaseAdapterError
+class GadflyAdapterCursor(ZopeCursor):
+
+ def executemany(self, operation, parameters):
+ command = operation.split(None, 1)[0].lower()
+ if command not in ("insert", "update", "delete"):
+ raise DatabaseAdapterError(
+ "executemany() is not applicable for %r" % operation)
+
+ operation, parameters = self._prepareOperation(operation, parameters)
+ self.connection.registerForTxn()
+ if command == "insert":
+ self.execute(operation, parameters)
+ else:
+ for param in parameters:
+ self.execute(operation, param)
+
+class GadflyAdapterConnection(ZopeConnection):
+
+ def cursor(self):
+ return GadflyAdapterCursor(self.conn.cursor(), self)
+
class GadflyAdapter(ZopeDatabaseAdapter):
"""A Gadfly adapter for Zope3"""
# The registerable object needs to have a container
__name__ = __parent__ = None
-
+
def _connection_factory(self):
"""Create a Gadfly DBI connection based on the DSN.
@@ -59,6 +81,14 @@
return db
+ def connect(self):
+ if not self.isConnected():
+ try:
+ self._v_connection = GadflyAdapterConnection(
+ self._connection_factory(), self)
+ except gadfly.error, error:
+ raise DatabaseException(str(error))
+
_gadflyRoot = 'gadfly'
def setGadflyRoot(path='gadfly'):
@@ -66,5 +96,4 @@
_gadflyRoot = path
def getGadflyRoot():
- global _gadflyRoot
return _gadflyRoot
Modified: Zope3/trunk/src/zope/app/rdb/interfaces.py
===================================================================
--- Zope3/trunk/src/zope/app/rdb/interfaces.py 2005-09-09 08:40:13 UTC (rev 38418)
+++ Zope3/trunk/src/zope/app/rdb/interfaces.py 2005-09-09 11:13:11 UTC (rev 38419)
@@ -93,6 +93,9 @@
def __str__(self):
return self.message
+class DatabaseAdapterError(DatabaseException):
+ pass
+
arraysize = 1 # default constant, symbolic
class IDBICursor(Interface):
Modified: Zope3/trunk/src/zope/app/rdb/tests/test_gadflyadapter.py
===================================================================
--- Zope3/trunk/src/zope/app/rdb/tests/test_gadflyadapter.py 2005-09-09 08:40:13 UTC (rev 38418)
+++ Zope3/trunk/src/zope/app/rdb/tests/test_gadflyadapter.py 2005-09-09 11:13:11 UTC (rev 38419)
@@ -141,12 +141,82 @@
cursor = connection.cursor()
verifyObject(IZopeCursor, cursor)
+class GadflyCursorStub(object):
+ def __init__(self):
+ self.operations = []
+
+ def execute(self, operation, parameters=None):
+ self.operations.append((operation, parameters))
+
+class GadflyConnectionStub(object):
+
+ def cursor(self):
+ return GadflyCursorStub()
+
+ def commit(self):
+ pass
+
+ def rollback(self):
+ pass
+
+ def close(self):
+ pass
+
+class GadflyTestAdapter(GadflyAdapter):
+
+ def _connection_factory(self):
+ return GadflyConnectionStub()
+
+class GadflyAdapterTests(TestCase):
+
+ def setUp(self):
+ self.adapter = GadflyTestAdapter("dbi://")
+ self.connection = self.adapter()
+ self.cursor = self.connection.cursor()
+
+ def testBadExecutemanyOperations(self):
+ raises = self.assertRaises
+ for operation in [
+ "SELECT",
+ "CREATE",
+ "DROP",
+ ]:
+ raises(DatabaseAdapterError,
+ self.cursor.executemany, operation, [])
+
+ def testExecutemanyInsert(self):
+ operation = "INSERT INTO table(v1, v2) VALUES (?, ?)"
+ parameters = [(1, 2), (3, 4)]
+ self.cursor.executemany(operation, parameters)
+ self.failUnlessEqual([(operation, parameters)],
+ self.cursor.operations)
+
+ def testExecutemanyUpdate(self):
+ operation = "UPDATE table SET value=0 WHERE id=?"
+ parameters = [(1,), (2,)]
+ self.cursor.executemany(operation, parameters)
+ self.failUnlessEqual([
+ (operation, parameters[0]),
+ (operation, parameters[1]),
+ ], self.cursor.operations)
+
+ def testExecutemanyDelete(self):
+ operation = "DELETE FROM table WHERE id=?"
+ parameters = [(1,), (2,)]
+ self.cursor.executemany(operation, parameters)
+ self.failUnlessEqual([
+ (operation, parameters[0]),
+ (operation, parameters[1]),
+ ], self.cursor.operations)
+
+
def test_suite():
return TestSuite((
makeSuite(TestGadflyAdapter),
makeSuite(TestGadflyAdapterNew),
makeSuite(TestGadflyAdapterDefault),
+ makeSuite(GadflyAdapterTests),
))
if __name__=='__main__':
More information about the Zope3-Checkins
mailing list