-
-
Save prostoalex/8763862 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from cStringIO import StringIO | |
| from gzip import GzipFile | |
| import zlib | |
| def gzip_string(s, level=6): | |
| """Compress string using gzip. | |
| Default compression level is 6. | |
| """ | |
| zbuf = StringIO() | |
| zfile = GzipFile(mode='wb', compresslevel=level, fileobj=zbuf) | |
| zfile.write(s) | |
| zfile.close() | |
| return zbuf.getvalue() | |
| def gunzip_string(s): | |
| """Decompress string using gzip. | |
| See http://stackoverflow.com/questions/2695152/in-python-how-do-i-decode-gzip-encoding/2695466#2695466 | |
| """ | |
| return zlib.decompress(s, 16 + zlib.MAX_WBITS) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import functools | |
| import time | |
| def repr_func(f): | |
| """Attempt to get the most useful string representation of callable. | |
| """ | |
| name = getattr(f, 'func_name', '<unknown>') | |
| func_code = getattr(f, 'func_code', None) | |
| if func_code is not None: | |
| return u'{name}() @ {fc.co_filename}:{fc.co_firstlineno}'.format( | |
| name=name, | |
| fc=func_code) | |
| return repr(f) | |
| def retry(tries, exceptions=(Exception,), delay=0): | |
| """ | |
| Decorator for retrying a function if exception occurs | |
| tries -- num tries | |
| exceptions -- exceptions to catch | |
| delay -- wait between retries | |
| """ | |
| def wrapper(func): | |
| @functools.wraps(func) | |
| def wrapped(*args, **kwargs): | |
| n = tries # need to declare local variable to modify it | |
| while n > 0: | |
| n -= 1 | |
| try: | |
| return func(*args, **kwargs) | |
| except exceptions as e: | |
| log.error(u'retry: {f} {e}', f=repr_func(func), e=e) | |
| time.sleep(delay) | |
| if n == 0: | |
| raise | |
| return wrapped | |
| return wrapper |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment