Author: phd
Date: 2006-04-06 08:52:32 -0600 (Thu, 06 Apr 2006)
New Revision: 1685
Modified:
home/phd/SQLObject/paramstyles/sqlobject/col.py
home/phd/SQLObject/paramstyles/sqlobject/dbconnection.py
home/phd/SQLObject/paramstyles/sqlobject/joins.py
home/phd/SQLObject/paramstyles/sqlobject/tests/test_transactions.py
Log:
Merged patches from the revisions 1678:1684 from the trunk
Modified: home/phd/SQLObject/paramstyles/sqlobject/col.py
===================================================================
--- home/phd/SQLObject/paramstyles/sqlobject/col.py 2006-04-06 14:51:16 UTC (rev 1684)
+++ home/phd/SQLObject/paramstyles/sqlobject/col.py 2006-04-06 14:52:32 UTC (rev 1685)
@@ -727,6 +727,31 @@
kw['name'] = style.instanceAttrToIDAttr(kw['name'])
super(SOForeignKey, self).__init__(**kw)
+ def sqliteCreateSQL(self):
+ sql = SOKeyCol.sqliteCreateSQL(self)
+ other = findClass(self.foreignKey)
+ tName = other.sqlmeta.table
+ idName = other.sqlmeta.idName
+ if self.cascade is not None:
+ if self.cascade == 'null':
+ action = 'ON DELETE SET NULL'
+ elif self.cascade:
+ action = 'ON DELETE CASCADE'
+ else:
+ action = 'ON DELETE RESTRICT'
+ else:
+ action = ''
+ constraint = ('CONSTRAINT %(colName)s_exists '
+ #'FOREIGN KEY(%(colName)s) '
+ 'REFERENCES %(tName)s(%(idName)s) '
+ '%(action)s' %
+ {'tName': tName,
+ 'colName': self.dbName,
+ 'idName': idName,
+ 'action': action})
+ sql = ' '.join([sql, constraint])
+ return sql
+
def postgresCreateSQL(self):
sql = SOKeyCol.postgresCreateSQL(self)
return sql
Modified: home/phd/SQLObject/paramstyles/sqlobject/dbconnection.py
===================================================================
--- home/phd/SQLObject/paramstyles/sqlobject/dbconnection.py 2006-04-06 14:51:16 UTC (rev 1684)
+++ home/phd/SQLObject/paramstyles/sqlobject/dbconnection.py 2006-04-06 14:52:32 UTC (rev 1685)
@@ -839,7 +839,8 @@
if not self._deletedCache.has_key(cls):
self._deletedCache[cls] = []
self._deletedCache[cls].append(inst.id)
- return self._dbConnection._SO_delete(inst)
+ meth = new.instancemethod(self._dbConnection._SO_delete.im_func, self, self.__class__)
+ return meth(inst)
def commit(self, close=False):
if self._obsolete:
Modified: home/phd/SQLObject/paramstyles/sqlobject/joins.py
===================================================================
--- home/phd/SQLObject/paramstyles/sqlobject/joins.py 2006-04-06 14:51:16 UTC (rev 1684)
+++ home/phd/SQLObject/paramstyles/sqlobject/joins.py 2006-04-06 14:52:32 UTC (rev 1685)
@@ -154,7 +154,11 @@
class SOSQLMultipleJoin(SOMultipleJoin):
def performJoin(self, inst):
- results = self.otherClass.select(getattr(self.otherClass.q, self.soClass.sqlmeta.style.dbColumnToPythonAttr(self.joinColumn)) == inst.id)
+ if inst.sqlmeta._perConnection:
+ conn = inst._connection
+ else:
+ conn = None
+ results = self.otherClass.select(getattr(self.otherClass.q, self.soClass.sqlmeta.style.dbColumnToPythonAttr(self.joinColumn)) == inst.id, connection=conn)
if self.orderBy is NoDefault:
self.orderBy = self.otherClass.sqlmeta.defaultOrder
return results.orderBy(self.orderBy)
@@ -239,12 +243,17 @@
%(otherTable)s.%(otherID)s = %(interTable)s.%(otherCol)s and
%(interTable)s.%(joinCol)s = %(table)s.%(ID)s and
%(table)s.%(ID)s = %(idValue)s''' % options
+ if inst.sqlmeta._perConnection:
+ conn = inst._connection
+ else:
+ conn = None
results = self.otherClass.select(sqlbuilder.SQLConstant(clause),
clauseTables=(
options['table'],
options['otherTable'],
options['interTable'],
- )
+ ),
+ connection=conn
)
# TODO (michelts): apply orderBy on the selection
return results
Modified: home/phd/SQLObject/paramstyles/sqlobject/tests/test_transactions.py
===================================================================
--- home/phd/SQLObject/paramstyles/sqlobject/tests/test_transactions.py 2006-04-06 14:51:16 UTC (rev 1684)
+++ home/phd/SQLObject/paramstyles/sqlobject/tests/test_transactions.py 2006-04-06 14:52:32 UTC (rev 1685)
@@ -51,3 +51,17 @@
assert bOut.name == 'robert'
finally:
TestSOTrans._connection.autoCommit = True
+
+def test_transaction_delete():
+ if not supports('transactions'):
+ return
+ setupClass(TestSOTrans)
+ trans = TestSOTrans._connection.transaction()
+ try:
+ TestSOTrans(name='bob')
+ bIn = TestSOTrans.byName('bob', connection=trans)
+ bIn.destroySelf()
+ bOut = TestSOTrans.select(TestSOTrans.q.name=='bob')
+ assert bOut.count() == 1
+ finally:
+ TestSOTrans._connection.autoCommit = True
|
