Changeset - e9e7c40b4f1a
[Not reviewed]
beta
0 2 0
Marcin Kuzminski - 13 years ago 2012-08-22 12:09:49
marcin@python-works.com
added deque with maxlen for py2.5 compat
2 files changed with 128 insertions and 2 deletions:
0 comments (0 inline, 0 general)
rhodecode/lib/compat.py
Show inline comments
 
@@ -25,7 +25,7 @@
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 

	
 
import os
 
from rhodecode import __platform__, PLATFORM_WIN
 
from rhodecode import __platform__, PLATFORM_WIN, __py_version__
 

	
 
#==============================================================================
 
# json
 
@@ -404,3 +404,129 @@ try:
 
    from io import BytesIO
 
except ImportError:
 
    from cStringIO import StringIO as BytesIO
 

	
 

	
 
#==============================================================================
 
# deque
 
#==============================================================================
 

	
 
if __py_version__ >= (2, 6):
 
    from collections import deque
 
else:
 
    #need to implement our own deque with maxlen
 
    class deque(object):
 

	
 
        def __init__(self, iterable=(), maxlen=-1):
 
            if not hasattr(self, 'data'):
 
                self.left = self.right = 0
 
                self.data = {}
 
            self.maxlen = maxlen
 
            self.extend(iterable)
 

	
 
        def append(self, x):
 
            self.data[self.right] = x
 
            self.right += 1
 
            if self.maxlen != -1 and len(self) > self.maxlen:
 
                self.popleft()
 

	
 
        def appendleft(self, x):
 
            self.left -= 1
 
            self.data[self.left] = x
 
            if self.maxlen != -1 and len(self) > self.maxlen:
 
                self.pop()
 

	
 
        def pop(self):
 
            if self.left == self.right:
 
                raise IndexError('cannot pop from empty deque')
 
            self.right -= 1
 
            elem = self.data[self.right]
 
            del self.data[self.right]
 
            return elem
 

	
 
        def popleft(self):
 
            if self.left == self.right:
 
                raise IndexError('cannot pop from empty deque')
 
            elem = self.data[self.left]
 
            del self.data[self.left]
 
            self.left += 1
 
            return elem
 

	
 
        def clear(self):
 
            self.data.clear()
 
            self.left = self.right = 0
 

	
 
        def extend(self, iterable):
 
            for elem in iterable:
 
                self.append(elem)
 

	
 
        def extendleft(self, iterable):
 
            for elem in iterable:
 
                self.appendleft(elem)
 

	
 
        def rotate(self, n=1):
 
            if self:
 
                n %= len(self)
 
                for i in xrange(n):
 
                    self.appendleft(self.pop())
 

	
 
        def __getitem__(self, i):
 
            if i < 0:
 
                i += len(self)
 
            try:
 
                return self.data[i + self.left]
 
            except KeyError:
 
                raise IndexError
 

	
 
        def __setitem__(self, i, value):
 
            if i < 0:
 
                i += len(self)
 
            try:
 
                self.data[i + self.left] = value
 
            except KeyError:
 
                raise IndexError
 

	
 
        def __delitem__(self, i):
 
            size = len(self)
 
            if not (-size <= i < size):
 
                raise IndexError
 
            data = self.data
 
            if i < 0:
 
                i += size
 
            for j in xrange(self.left + i, self.right - 1):
 
                data[j] = data[j + 1]
 
            self.pop()
 

	
 
        def __len__(self):
 
            return self.right - self.left
 

	
 
        def __cmp__(self, other):
 
            if type(self) != type(other):
 
                return cmp(type(self), type(other))
 
            return cmp(list(self), list(other))
 

	
 
        def __repr__(self, _track=[]):
 
            if id(self) in _track:
 
                return '...'
 
            _track.append(id(self))
 
            r = 'deque(%r, maxlen=%s)' % (list(self), self.maxlen)
 
            _track.remove(id(self))
 
            return r
 

	
 
        def __getstate__(self):
 
            return (tuple(self),)
 

	
 
        def __setstate__(self, s):
 
            self.__init__(s[0])
 

	
 
        def __hash__(self):
 
            raise TypeError
 

	
 
        def __copy__(self):
 
            return self.__class__(self)
 

	
 
        def __deepcopy__(self, memo={}):
 
            from copy import deepcopy
 
            result = self.__class__()
 
            memo[id(self)] = result
 
            result.__init__(deepcopy(tuple(self), memo))
 
            return result
rhodecode/lib/subprocessio.py
Show inline comments
 
@@ -25,7 +25,7 @@ If not, see <http://www.gnu.org/licenses
 
import os
 
import subprocess
 
import threading
 
from collections import deque
 
from rhodecode.lib.compat import deque
 

	
 

	
 
class StreamFeeder(threading.Thread):
0 comments (0 inline, 0 general)