Python共通モジュールのコレクション

Python共通モジュールのコレクション

一般的なモジュールは、主に次のカテゴリに分類されます（不足しているモジュールは後で追加されます）。

• 時間変換

• 時間計算

• シリアル化と逆シリアル化： json、 pickle

• コーデック： unicode、 base64

• 暗号化と復号化： md5、 sha1、 hmac_sha1、 aes

• 一般的なデコレータ：

• 実行時間デコレータを計算する

• キャッシュデコレータ

• デコレータの再試行エラー

• 遅延デコレータ

• テール再帰最適化デコレータ

• ini構成ファイルの読み取り

コードは次のように統合されています。

#! /usr/bin/env python
# - *- coding: utf-8-*-"""
Created on 9/21/171:46 PM
@ author: Chen Liang
@ function:pythonで一般的に使用されるモジュールのコレクション、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の一般的な単位：分類するのは簡単ではありませんが、一般的に使用される方法がここに追加されます"""
 pass

classTimeTransferUtil(object):"""時間に関連する一般的な変換方法"""classTimeUtil(object):"""時間に関連する一般的な計算方法"""
 @ staticmethod
 def str_to_date():
  pass

classSerializeUtil(object):"""シリアル化と逆シリアル化：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):"""エンコードとデコードに関連する一般的な方法：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):"""暗号化と復号化に関連する一般的な方法：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暗号化
  : param s:暗号化する文字列
  : param key:キー
  : param salt:塩, 16bit eg. b'0000000101000000':param mode:AESモード
  : return:暗号化された文字列
        """
  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復号化
  : param s:復号化する文字列
  : param key:キー
  : param salt:塩, 16bit eg. b'0000000101000000':param mode:AESモード
  : return:復号化された文字列
        """
  cipher = AES.new(hashlib.md5(key).hexdigest(), mode, salt)return cipher.decrypt(a2b_hex(s)).rstrip('\0')classTailRecurseException:"""テール再帰例外"""
 def __init__(self, args, kwargs):
  self.args = args
  self.kwargs = kwargs

classDecoratorUtil(object):"""一般的なデコレータ：実行時間timeit、キャッシュキャッシュ、エラー再試行再試行"""

 __ cache_dict ={}

 @ staticmethod
 def timeit(fn):"""計算実行時間"""
  @ 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):"""期限切れ"""if duration ==-1:return False
  return time.time()- entry['time']> duration

 @ staticmethod
 def __compute_key(fn, args, kw):"""シリアル化してハッシュ"""
  key = pickle.dumps((fn.__name__, args, kw))return hashlib.sha1(key).hexdigest()

 @ classmethod
 def cache(cls, expired_time=-1):"""
  キャッシュ
  : param expired_time:有効期限、-1は有効期限がないことを意味します
  : return:キャッシュされた結果または計算された結果を返す
        """
  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):"""
  エラーの再試行
  : param exceptions:ValueErrorなどの単一の例外,またはタプル,タプル要素は異常です。(ValueError, TypeError):param retry_times:再試行回数
  : param time_pause:初期休止時間
  : param time_offset:一時停止時間のオフセット倍数。デフォルトではオフセットなし
  : return:成功の値を返すか、試行回数を超えたときに例外をスローします
        """
  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):"""
  遅延デコレータは、関数の実行前後の遅延の追加をサポートしています。関数の前後に追加する場合は、2つの装飾を使用できます。
  time.スリープは現在のスレッドのみをブロックし、プロセス全体をブロックすることはなく、他のスレッドは影響を受けません
  : param delay_time:遅延時間、フロートタイプ
  : param mode:モード。関数の実行前または実行後に遅延を追加するかどうかを指定します。値はBEFOREです。(1)または後で(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):"""テール再帰最適化デコレータ、デコレーションされた関数がテール再帰関数でない場合、エラーが報告されます"""
  @ 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構成ファイルの読み取り"""
 def __init__(self,*file_names):"""
  init
  : param file_names:複数の要素を含む反復可能なオブジェクト
        """
  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))

不足している部分は後で追加されます。穴を埋めることを忘れないでください。