forked from celery/celery
-
Notifications
You must be signed in to change notification settings - Fork 40
/
datastructures.py
72 lines (46 loc) · 1.74 KB
/
datastructures.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
"""
Custom Datastructures
"""
from UserList import UserList
import traceback
class PositionQueue(UserList):
"""A positional queue of a specific length, with slots that are either
filled or unfilled. When all of the positions are filled, the queue
is considered :meth:`full`.
:param length: see :attr:`length`.
.. attribute:: length
The number of items required for the queue to be considered full.
"""
class UnfilledPosition(object):
"""Describes an unfilled slot."""
def __init__(self, position):
self.position = position
def __init__(self, length):
self.length = length
self.data = map(self.UnfilledPosition, xrange(length))
def full(self):
"""Returns ``True`` if all of the slots has been filled."""
return len(self) >= self.length
def __len__(self):
"""``len(self)`` -> number of slots filled with real values."""
return len(self.filled)
@property
def filled(self):
"""Returns the filled slots as a list."""
return filter(lambda v: not isinstance(v, self.UnfilledPosition),
self.data)
class ExceptionInfo(object):
"""Exception wrapping an exception and its traceback.
:param exc_info: The exception tuple info as returned by
:func:`traceback.format_exception`.
.. attribute:: exception
The original exception.
.. attribute:: traceback
A traceback from the point when :attr:`exception` was raised.
"""
def __init__(self, exc_info):
type_, exception, tb = exc_info
self.exception = exception
self.traceback = '\n'.join(traceback.format_exception(*exc_info))
def __str__(self):
return str(self.exception)