本文整理汇总了Python中transaction.weakset.WeakSet类的典型用法代码示例。如果您正苦于以下问题：Python WeakSet类的具体用法？Python WeakSet怎么用？Python WeakSet使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了WeakSet类的13个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_map_w_gced_element
def test_map_w_gced_element(self):
import gc
from transaction.weakset import WeakSet
w = WeakSet()
dummy = Dummy()
dummy2 = Dummy()
dummy3 = [Dummy()]
w.add(dummy)
w.add(dummy2)
w.add(dummy3[0])
_orig = w.as_weakref_list
def _as_weakref_list():
# simulate race condition during iteration of list
# object is collected after being iterated.
result = _orig()
del dummy3[:]
gc.collect()
return result
w.as_weakref_list = _as_weakref_list
def poker(x):
x.poked = 1
w.map(poker)
for thing in dummy, dummy2:
self.assertEqual(thing.poked, 1)
开发者ID:hughy603，项目名称:pyramid_tutorial，代码行数:26，代码来源:test_weakset.py
示例2: test_contains
def test_contains(self):
from transaction.weakset import WeakSet
w = WeakSet()
dummy = Dummy()
w.add(dummy)
self.assertEqual(dummy in w, True)
dummy2 = Dummy()
self.assertEqual(dummy2 in w, False)
开发者ID:hughy603，项目名称:pyramid_tutorial，代码行数:8，代码来源:test_weakset.py
示例3: test_remove
def test_remove(self):
from transaction.weakset import WeakSet
w = WeakSet()
dummy = Dummy()
w.add(dummy)
self.assertEqual(dummy in w, True)
w.remove(dummy)
self.assertEqual(dummy in w, False)
开发者ID:hughy603，项目名称:pyramid_tutorial，代码行数:8，代码来源:test_weakset.py
示例4: test_len
def test_len(self):
import gc
w = WeakSet()
d1 = Dummy()
d2 = Dummy()
w.add(d1)
w.add(d2)
self.assertEqual(len(w), 2)
del d1
gc.collect()
self.assertEqual(len(w), 1)
开发者ID:mbbui，项目名称:Jminee，代码行数:12，代码来源:test_weakset.py
示例5: test_len
def test_len(self):
import gc
from transaction.weakset import WeakSet
w = WeakSet()
d1 = Dummy()
d2 = Dummy()
w.add(d1)
w.add(d2)
self.assertEqual(len(w), 2)
del d1
gc.collect()
self.assertEqual(len(w), 1)
开发者ID:jean，项目名称:transaction，代码行数:12，代码来源:test_weakset.py
示例6: test_len
def test_len(self):
import gc
from transaction.weakset import WeakSet
w = WeakSet()
d1 = Dummy()
d2 = Dummy()
w.add(d1)
w.add(d2)
self.assertEqual(len(w), 2)
del d1
gc.collect()
if not JYTHON:
# The Jython GC is non deterministic
self.assertEqual(len(w), 1)
开发者ID:hughy603，项目名称:pyramid_tutorial，代码行数:14，代码来源:test_weakset.py
示例7: TransactionManager
class TransactionManager(object):
def __init__(self):
self._txn = None
self._synchs = WeakSet()
def begin(self):
if self._txn is not None:
self._txn.abort()
txn = self._txn = Transaction(self._synchs, self)
_new_transaction(txn, self._synchs)
return txn
def get(self):
if self._txn is None:
self._txn = Transaction(self._synchs, self)
return self._txn
def free(self, txn):
assert txn is self._txn
self._txn = None
def registerSynch(self, synch):
self._synchs.add(synch)
def unregisterSynch(self, synch):
self._synchs.remove(synch)
def isDoomed(self):
return self.get().isDoomed()
def doom(self):
return self.get().doom()
def commit(self):
return self.get().commit()
def abort(self):
return self.get().abort()
def savepoint(self, optimistic=False):
return self.get().savepoint(optimistic)
开发者ID:zerodivisionfeature，项目名称:pyCare，代码行数:42，代码来源:_manager.py
示例8: test_map
def test_map(self):
from transaction.weakset import WeakSet
w = WeakSet()
dummy = Dummy()
dummy2 = Dummy()
dummy3 = Dummy()
w.add(dummy)
w.add(dummy2)
w.add(dummy3)
def poker(x):
x.poked = 1
w.map(poker)
for thing in dummy, dummy2, dummy3:
self.assertEqual(thing.poked, 1)
开发者ID:hughy603，项目名称:pyramid_tutorial，代码行数:14，代码来源:test_weakset.py
示例9: test_as_weakref_list
def test_as_weakref_list(self):
w = WeakSet()
dummy = Dummy()
dummy2 = Dummy()
dummy3 = Dummy()
w.add(dummy)
w.add(dummy2)
w.add(dummy3)
del dummy3
L = [x() for x in w.as_weakref_list()]
# L is a list, but it does not have a guaranteed order.
self.assertTrue(list, type(L))
self.assertEqual(set(L), set([dummy, dummy2]))
开发者ID:c0ns0le，项目名称:zenoss-4，代码行数:13，代码来源:test_weakset.py
示例10: test_as_weakref_list
def test_as_weakref_list(self):
import gc
from transaction.weakset import WeakSet
w = WeakSet()
dummy = Dummy()
dummy2 = Dummy()
dummy3 = Dummy()
w.add(dummy)
w.add(dummy2)
w.add(dummy3)
del dummy3
gc.collect()
L = [x() for x in w.as_weakref_list()]
# L is a list, but it does not have a guaranteed order.
self.assertTrue(list, type(L))
self.assertEqual(set(L), set([dummy, dummy2]))
开发者ID:serg0987，项目名称:python，代码行数:16，代码来源:test_weakset.py
示例11: TransactionManager
class TransactionManager(object):
def __init__(self):
self._txn = None
self._synchs = WeakSet()
def begin(self):
if self._txn is not None:
self._txn.abort()
txn = self._txn = Transaction(self._synchs, self)
_new_transaction(txn, self._synchs)
return txn
__enter__ = lambda self: self.begin()
def get(self):
if self._txn is None:
self._txn = Transaction(self._synchs, self)
return self._txn
def free(self, txn):
assert txn is self._txn
self._txn = None
def registerSynch(self, synch):
self._synchs.add(synch)
def unregisterSynch(self, synch):
self._synchs.remove(synch)
def isDoomed(self):
return self.get().isDoomed()
def doom(self):
return self.get().doom()
def commit(self):
return self.get().commit()
def abort(self):
return self.get().abort()
def __exit__(self, t, v, tb):
if v is None:
self.commit()
else:
self.abort()
def savepoint(self, optimistic=False):
return self.get().savepoint(optimistic)
def attempts(self, number=3):
assert number > 0
while number:
number -= 1
if number:
yield Attempt(self)
else:
yield self
def _retryable(self, error_type, error):
if issubclass(error_type, TransientError):
return True
for dm in self.get()._resources:
should_retry = getattr(dm, 'should_retry', None)
if (should_retry is not None) and should_retry(error):
return True
开发者ID:Andyvs，项目名称:TrackMonthlyExpenses，代码行数:68，代码来源:_manager.py
示例12: __init__
def __init__(self):
self._txn = None
self._synchs = WeakSet()
开发者ID:hughy603，项目名称:pyramid_tutorial，代码行数:3，代码来源:_manager.py
示例13: TransactionManager
class TransactionManager(object):
def __init__(self):
self._txn = None
self._synchs = WeakSet()
def begin(self):
""" See ITransactionManager.
"""
if self._txn is not None:
self._txn.abort()
txn = self._txn = Transaction(self._synchs, self)
_new_transaction(txn, self._synchs)
return txn
__enter__ = lambda self: self.begin()
def get(self):
""" See ITransactionManager.
"""
if self._txn is None:
self._txn = Transaction(self._synchs, self)
return self._txn
def free(self, txn):
if txn is not self._txn:
raise ValueError("Foreign transaction")
self._txn = None
def registerSynch(self, synch):
""" See ITransactionManager.
"""
self._synchs.add(synch)
if self._txn is not None:
synch.newTransaction(self._txn)
def unregisterSynch(self, synch):
""" See ITransactionManager.
"""
self._synchs.remove(synch)
def clearSynchs(self):
""" See ITransactionManager.
"""
self._synchs.clear()
def registeredSynchs(self):
""" See ITransactionManager.
"""
return bool(self._synchs)
def isDoomed(self):
""" See ITransactionManager.
"""
return self.get().isDoomed()
def doom(self):
""" See ITransactionManager.
"""
return self.get().doom()
def commit(self):
""" See ITransactionManager.
"""
return self.get().commit()
def abort(self):
""" See ITransactionManager.
"""
return self.get().abort()
def __exit__(self, t, v, tb):
if v is None:
self.commit()
else:
self.abort()
def savepoint(self, optimistic=False):
""" See ITransactionManager.
"""
return self.get().savepoint(optimistic)
def attempts(self, number=3):
if number <= 0:
raise ValueError("number must be positive")
while number:
number -= 1
if number:
yield Attempt(self)
else:
yield self
def _retryable(self, error_type, error):
if issubclass(error_type, TransientError):
return True
for dm in self.get()._resources:
should_retry = getattr(dm, 'should_retry', None)
if (should_retry is not None) and should_retry(error):
return True
开发者ID:hughy603，项目名称:pyramid_tutorial，代码行数:100，代码来源:_manager.py
注：本文中的transaction.weakset.WeakSet类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论