-
-
Save aburan28/8762591 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
| def str_utf8(x): | |
| """ | |
| Returns the byte string representation of obj. | |
| Like unicode(x).encode('utf-8') except it works for bytes. | |
| """ | |
| if isinstance(x, str): | |
| return x | |
| return unicode(x).encode('utf-8') |
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 logbook | |
| import logging | |
| import time | |
| # log = logbook.Logger(__name__) | |
| log = logging.getLogger(__name__) | |
| 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 # copy to local variable for modification | |
| while n > 0: | |
| n -= 1 | |
| try: | |
| return func(*args, **kwargs) | |
| except exceptions as e: | |
| if n == 0: | |
| raise | |
| # logbook | |
| # log.error(u'retry: {f} {e}', f=repr_func(func), e=e) | |
| # logging | |
| log.error(u'retry: %s %s', repr_func(func), e) | |
| time.sleep(delay) | |
| return wrapped | |
| return wrapper |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment