Collection of Python Common Modules

Collection of Python Common Modules

Common modules are mainly divided into the following categories (the missing ones will be added later):

• Time conversion

• Time calculation

• Serialization and deserialization: json, pickle

• Codec: unicode, base64

• Encryption and decryption: md5, sha1, hmac_sha1, aes

• Common decorators:

• Calculate execution time decorator

• Cache decorator

• Error retry decorator

• Delay decorator

• Tail recursive optimization decorator

• ini configuration file read

The code is integrated as follows:

#! /usr/bin/env python
# - *- coding: utf-8-*-"""
Created on 9/21/171:46 PM
@ author: Chen Liang
@ function:Collection of python commonly used modules, util.py
"""

import time
import datetime
import ConfigParser
import ast
import sys
import json
import pickle
import base64
import hashlib
from Crypto.Cipher import AES
from binascii import b2a_hex, a2b_hex
from functools import wraps

BEFORE =1
LATER =2classCommonUtil(object):"""Python general unit: it is not easy to categorize but the commonly used method is added here"""
 pass

classTimeTransferUtil(object):"""Time-related common conversion methods"""classTimeUtil(object):"""Time-related common calculation methods"""
 @ staticmethod
 def str_to_date():
  pass

classSerializeUtil(object):"""Serialization and deserialization: json, pickle"""
 @ staticmethod
 def json_loads(json_str, encoding=None):try:
   obj = json.loads(s=json_str, encoding=encoding)return True, obj
  except ValueError as e:return False,str(e)
  except Exception as e:return False,str(e)

 @ staticmethod
 def json_dumps(obj):try:
   json_str = json.dumps(obj=obj)return True, json_str
  except TypeError as e:return False,str(e)
  except Exception as e:return False,str(e)

 @ staticmethod
 def pickle_loads(pickle_str):try:
   obj = pickle.loads(pickle_str)return True, obj
  except IndexError as e:return False,str(e)
  except Exception as e:return False,str(e)

 @ staticmethod
 def pickle_dumps(obj):try:
   pickle_str = pickle.dumps(obj)return True, pickle_str
  except Exception as e:return False,str(e)classCodecUtil(object):"""Common methods related to encoding and decoding: base64 unicode"""
 @ staticmethod
 def base64_encode(data):try:return True, base64.b64encode(data)
  except TypeError as e:return False,str(e)
  except Exception as e:return False,str(e)

 @ staticmethod
 def base64_decode(encoded_data):try:return True, base64.b64decode(encoded_data)
  except TypeError as e:return False,str(e)
  except Exception as e:return False,str(e)

 @ staticmethod
 def to_unicode(s, encoding='utf-8'):return s ifisinstance(s, unicode)elseunicode(s, encoding)

 @ staticmethod
 def unicode_to(unicode_s, encoding='utf-8'):return unicode_s.encode(encoding)classCryptoUtil(object):"""Common methods related to encryption and decryption: md5 aes"""
 @ staticmethod
 def md5(str_object):"""md5"""
  m = hashlib.md5()
  m.update(str_object)return m.hexdigest()

 @ staticmethod
 def aes_encrypt(s, key, salt, mode=AES.MODE_CBC):"""
  aes encryption
  : param s:String to be encrypted
  : param key:Key
  : param salt:salt, 16bit eg. b'0000000101000000':param mode:AES mode
  : return:Encrypted string
        """
  cipher = AES.new(hashlib.md5(key).hexdigest(), mode, salt)
  n_text = s +('\0'*(16-(len(s)%16)))returnb2a_hex(cipher.encrypt(n_text))

 @ staticmethod
 def aes_decrypt(s, key, salt, mode=AES.MODE_CBC):"""
  aes decryption
  : param s:String to be decrypted
  : param key:Key
  : param salt:salt, 16bit eg. b'0000000101000000':param mode:AES mode
  : return:The decrypted string
        """
  cipher = AES.new(hashlib.md5(key).hexdigest(), mode, salt)return cipher.decrypt(a2b_hex(s)).rstrip('\0')classTailRecurseException:"""Tail recursion exception"""
 def __init__(self, args, kwargs):
  self.args = args
  self.kwargs = kwargs

classDecoratorUtil(object):"""Common decorators: execution time timeit, cache cache, error retry retry"""

 __ cache_dict ={}

 @ staticmethod
 def timeit(fn):"""Calculation execution time"""
  @ wraps(fn)
  def wrap(*args,**kwargs):
   start = time.time()
   ret =fn(*args,**kwargs)
   end = time.time()
   print "@timeit: {0} tasks, {1} secs".format(fn.__name__,str(end - start))return ret
  return wrap

 @ staticmethod
 def __is_expired(entry, duration):"""Expired"""if duration ==-1:return False
  return time.time()- entry['time']> duration

 @ staticmethod
 def __compute_key(fn, args, kw):"""Serialize and hash"""
  key = pickle.dumps((fn.__name__, args, kw))return hashlib.sha1(key).hexdigest()

 @ classmethod
 def cache(cls, expired_time=-1):"""
  Cache
  : param expired_time:Expiration,-1 means no expiration
  : return:Return cached results or calculated results
        """
  def _cache(fn):
   @ wraps(fn)
   def wrap(*args,**kwargs):
    key = cls.__compute_key(fn, args, kwargs)if key in cls.__cache_dict:if cls.__is_expired(cls.__cache_dict[key], expired_time) is False:return cls.__cache_dict[key]['value']
    ret =fn(*args,**kwargs)
    cls.__cache_dict[key]={'value': ret,'time': time.time()}return ret
   return wrap
  return _cache

 @ staticmethod
 def retry(exceptions, retry_times=3, time_pause=3, time_offset=1):"""
  Error retry
  : param exceptions:Single exception such as ValueError,Or tuple,Tuple elements are abnormal, such as(ValueError, TypeError):param retry_times:number of retries
  : param time_pause:Initial pause time
  : param time_offset:The offset multiple of the pause time, no offset by default
  : return:Return a successful value, or throw an exception when the number of attempts is over
        """
  def _retry(fn):
   @ wraps(fn)
   def wrap(*args,**kwargs):
    retry_times_tmp, time_pause_tmp = retry_times, time_pause
    while retry_times_tmp >1:try:returnfn(*args,**kwargs)
     except exceptions:
      time.sleep(time_pause_tmp)
      retry_times_tmp -=1
      time_pause_tmp *= time_offset
    returnfn(*args,**kwargs)return wrap
  return _retry

 @ staticmethod
 def delay(delay_time=3, mode=BEFORE):"""
  Delay decorator supports adding delay before and after the function execution. If you want to add before and after the function, you can use two decorations.
  time.Sleep will only block the current thread and will not block the entire process, and other threads will not be affected
  : param delay_time:Delay time, float type
  : param mode:Mode, specify whether to add a delay before the function execution or after the execution, the value is BEFORE(1)Or LATER(2):return:"""
  def _delay(fn):
   @ wraps(fn)
   def wrap(*args,**kwargs):if mode == BEFORE:
     time.sleep(delay_time)
    ret =fn(*args,**kwargs)if mode == LATER:
     time.sleep(delay_time)return ret
   return wrap
  return _delay

 @ staticmethod
 def tail_call_optimized(fn):"""Tail recursive optimization decorator, if the decorated function is not a tail recursive function, an error will be reported"""
  @ wraps(fn)
  def wrap(*args,**kwargs):
   f = sys._getframe()if f.f_back and f.f_back.f_back and f.f_back.f_back.f_code == f.f_code:
    raise TailRecurseException(args, kwargs)else:while True:try:returnfn(*args,**kwargs)
     except TailRecurseException as e:
      args = e.args
      kwargs = e.kwargs
  return wrap

classIniConfigParserUtil(object):"""ini configuration file read"""
 def __init__(self,*file_names):"""
  init
  : param file_names:Iterable object containing multiple elements
        """
  self.config = ConfigParser.ConfigParser()for file_name in file_names:try:
    self.config.readfp(open(file_name,'rb'))break
   except IOError:continueelse:
   sys.exit('All files have failed to read')

 def get_string(self, section, option):return self.config.get(section, option)

 def get_int(self, section, option):return self.config.getint(section, option)

 def get_float(self, section, option):return self.config.getfloat(section, option)

 def get_boolean(self, section, option):return self.config.getboolean(section, option)

 def get_list(self, section, option):return ast.literal_eval(self.config.get(section, option))

 def get_dict(self, section, option):return ast.literal_eval(self.config.get(section, option))

The missing part will be added later, remember to fill in the hole.