加入收藏 | 设为首页 |

不稂不莠-回绝重复造轮子!python实用工具类及函数大引荐!

海外新闻 时间: 浏览:177 次

从2015年转行程序员至今也有两年多了,最初自学的java却误打误撞进了python的坑,入职之后一天java也没有写过,或许这能够了解成缘分吧,哈哈!运用python作业久了,随手写一些小东西再所不免,不知不觉,我的东西包也增加到了几千行代码。其间有的函数按照他人的代码改写,也有的源于自己的创意,我自以为仍是挺好用的。现在一致规整后做成pip包,拿出来共享给咱们运用。

函数都很细巧,代码完成相对简略,风格比较自在,尽或许不依靠其它装置包,极个别的东西依靠一些特定的驱动程序,比方redis,kafka, psutil,假如自己的项目没有相关需求,那么能够将需求的办法,类仿制放到自己的项目中,不用完全装置ShichaoMa/toolkit。定心,我是不会追查版权的。

正文开端

装置toolkity

#python3测验有用
pip install toolkity

有朋友或许会想为什么不叫toolkit,我,也是那么想的。可是姓名被他人占用了肿么办,只能在后面加个y,看起来正式而不失一点小幽默。

装置结束,来跟我学习几个常用函数的运用办法吧。

  • Timer 简略好用的计时器
class Timer(object):
"""
计时器,关于需求计时的代码进行with操作:
with Timer() as timer:
...
...
print(timer.cost)
...
"""
def __init__(self, start=None):不稂不莠-回绝重复造轮子!python实用工具类及函数大引荐!
self.start = start if start is not None else time.time()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.stop = time.time()
self.cost = self.stop - self.start
return exc_type is None

有时咱们想对咱们写的代码执行时刻进行计算,一般咱们会运用如下代码:

import time
start = time.time()
...# 咱们的代码
end = time.time()
cost = end - start

现在有了timer,一切都变的简略啦

# 运用办法:
In [2]: from toolkit.managers import Timer
In [3]: with Timer() as timer:
...: num = 10**20
...:
In [4]: print(timer.cost)
3.6716461181640625e-05
In [5]: print(timer.start)
1512270309.2823365
In [6]: print(timer.stop)
1512270309.2823732

一起,你还能够指定开端时刻

In [9]: import time
In [10]: with Timer(start=time.time()-10) as timer:
...: num = 12**12
...:
In [11]: timer.cost
Out[11]: 10.000059366226196
  • ExceptContext 反常捕获上下文
class ExceptContext(object):
"""
反常捕获上下文
eg:
def test():
with ExceptContext(Exception, errback=lambda name, *args:print(name)):
raise Exception("test. ..")
"""
def __init__(self, exception=Exception, func_name=None,
errback=lambda func_name, *args: traceback.print_exception(*args) is None,
finalback=lambda got_err: got_err):
"""
:param exception: 指定要监控的反常
:param func_name: 能够挑选供给当时地点函数的称号,回调函数会提交到函数,用于盯梢
:param errback: 供给一个回调函数,假如发作了指定反常,就调用该函数,该函数的回来值为True时不会继续抛出反常
:param finalback: finally要做的操作
"""
self.errback = errback
self.finalback = finalback
self.exception = exception
self.got_err = False
self.func_name = func_name or _find_caller_name(is_func=True)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
return_code = False
if isinstance(exc_val, self.exception):
self.got_err = True
return_code = self.errback(self.func_name, exc_type, exc_val, exc_tb)
self.finalback(self.got_err)
return return_code

一般咱们想捕获反常时,会写如下代码

got_err = False
try:
1/0
except ZeroDivisionError:
import traceback
traceback.print_exc()
got_err = True
finally:
print(got_err)

现在咱们能够用ExceptContext简略完成

with ExceptContext(ZeroDivisionError, finalback=lambda got_err: print(got_err)):
1/0

其间ExceptContext接纳4个参数

  1. :param exception: 指定要监控的反常, 默以为Exception:param func_name: 能够挑选供给当时地点函数的称号,回调函数会提交到函数,用于盯梢,默以为None,自己判别调用函数称号:param errback: 供给一个回调函数,假如发作了指定反常,就调用该函数,该函数的回来值为True时不会继续抛出反常 默许打出反常信息,回来True:param finalback: finally要做的操作 默许回来是否发作反常。

经过自定义errback,咱们能够对反常做任何想要的操作。

  • debugger debug小东西
def debugger():
try:
debug = eval(os.environ.get("DEBUG"))
except Exception:
debug = False
if debug:
d = pdb.Pdb()
d.set_trace( sys._getframe().f_back)

有时咱们或许会运用pdb来调试,常常会发作的状况是,咱们在测验环境下调试,布置到出产之后,发现pdb.set_trace()忘掉删去,导致代码在出产体系中卡住,这就很为难了。运用debugger,一起在测验环境中参加export DEBUG=True,能够轻松防止上述状况

import os
os.env不稂不莠-回绝重复造轮子!python实用工具类及函数大引荐!iron["DEBUG"] = True
from toolkit import debugger
def fun():
debugger()
....
fun()
  • duplicate 保序去重函数
def duplicate(iterable, keep=lambda x: x, key=lambda x: x, reverse=False):
"""
保序去重
:param iterable:
:param keep: 去重的一起要对element做的操作
:param key: 运用哪一部分去重
:param reverse: 是否反向去重
:return:
"""
result = list()
duplicator = list()
if reverse:
iterable = reversed(iterable)
for i in iterable:
keep_field = keep(i)
key_words = key(i)
if key_words not in duplicator:
result.append(keep_field)
duplicator.append(key_words)
return list(reversed(result)) if reverse else result

咱们常常会遇到去重问题,比方

In [2]: ls = [3,4,5,2,4,1]
In [3]: ls = list(set(ls))
In [4]: ls
Out[4]: [1, 2, 3, 4, 5]

上述列表中有2个4,咱们想去掉剩余的4,可是不想次序乱掉,假如运用list(set(ls))的办法,次序会乱掉,因而咱们能够运用duplicate函数来做

In [5]: from toolkit import duplicate
In [6]: ls = [3, 4, 5, 2, 4, 1]
# 正序去重
In [7]: duplicate(ls)
Out[7]: [3, 4, 5, 2, 1]
# 逆序去重
In [8]: duplicate(ls, reverse=True)
Out[8]: [3, 5, 2, 4, 1]
# 指定规矩去重
In [10]: ls = [{"a": 3, "b": 4}, {"a":3, "b": 5}]
In [11]: duplicate(ls, key=lambda x: x["a"])
Out[11]: [{'a': 3, 'b': 4}]
# 去重后仅保存部分数据
In [12]: duplicate(ls, key=lambda x: x["a"], keep=lambda x: x["b"])
Out[12]: [4]
  • chain_all 衔接多个可迭代目标
def chain_all(iter):
"""
衔接多个序列或字典
:param iter:
:return:
"""
iter = list(iter)
if not iter:
return []
if isinstance(iter[0], dict):
result = {}
for i in iter:
result.update(i)
else:
result = reduce(lambda x, y: list(x)+list(y), iter)
return result

曾经咱们都运用Itertools.chain.from_iterable,可是这个函数无法衔接字典一起回来的是可迭代器目标。

In [14]: chain_all([[1, 2], [1, 2]])
Out[14]: [1, 2, 1, 2]
In [15]: chain_all([{"a": 1}, {"b": 2}])
Out[15]: {'a': 1, 'b': 2}
  • safely_json_loads 安全的将字符串变成json目标
# 代码完成
def safely_json_loads(json_str, defaulttype=dict, escape=True):
"""
回来安全的json类型
:param json_str: 要被loads的字符串
:param defaulttype: 若load失利期望得到的目标类型
:param escape: 是否将单引号变成双引号
:return:
"""
if not json_str:
return defaulttype()
elif escape:
data = replace_quote(json_str)
return json.loads(data)
else:
return json.loads(json_str)

关于空字符串,回来默许类型,关于运用单引号包裹的字符串,将其转换成双引号

In [17]: json_str = """{'a': 1, "b": "i'm tom."}"""
In [18]: safely_json_loads(json_str)
Out[18]: {'a': 1, 'b': "i'm tom."}
# 上面的i'm tom中的单引号不会被转成双引号
  • format_html_string 格式化html
def format_html_string(html):
"""
格式化html, 去掉剩余的字符,类,script等。
:param html:
:return:
"""
trims = [(r'\n',''),
(r'\t', ''),
(r'\r', ''),
(r' ', ''),
(r'\\u2018', 不稂不莠-回绝重复造轮子!python实用工具类及函数大引荐!"'"),
(r'\\u2019', "'"),
(r'\\ufeff', ''),
(r'\\u2022', ":"),
(r"<([a-z][a-z0-9]*)\ [^>]*>", '<\g<1>>'),
(r'<\s*script[^>]*>[^<]*<\s*/\s*script\s*>', ''),
(r"", '')]
return reduce(lambda string, replacement: re.sub(replacement[0], replacement[1], string), trims, html)

去掉剩余的html特点

In [20]: format_html_string("
")
Out[20]: '
'

....

除以上东西以外,还有许多东西十分有用,由于比如相对杂乱,就不逐个举例了,能够多重视我的github, 如ShichaoMa/proxy_factory项目,就有许多用例呈现。

下面简略介绍一些其它东西

  • 重试器,包装函数对指定反常进行重试
def retry_wrapper(retry_times, exception=Exception, error_handler=None, interval=0.1):
"""
函数重试装修器
:param retry_times: 重试次数
:param exception: 需求重试的反常
:param error_handler: 犯错时的回调函数
:param interval: 重试距离时刻
:return:
"""
def out_wrapper(func):
@wraps(func)
def wrapper(*args, **kwargs):
count = 0
while True:
try:
return func(*args, **kwargs)
except exception as e:
count += 1
if error_handler:
result = error_handler(func.__name__, count, e, *args, **kwargs)
if result:
count -= 1
if count >= retry_times:
raise
time.sleep(interval)
return wrappe不稂不莠-回绝重复造轮子!python实用工具类及函数大引荐!r
return out_wrapper
  • 超时器,装修函数并指定其超时时刻
def timeout(timeout_time, default):
"""
Decorate a method so it is required to execute in a given time period,
or return a default value.
:param timeout_time:
:param default:
:return:
"""
class DecoratorTimeout(Exception):
pass
def timeout_function(f):
def f2(*args):
def timeout_handler(signum, frame):
raise DecoratorTimeout()
old_handler = signal.signal(signal.SIGALRM, timeout_handler)
# triger alarm in timeout_time seconds
signal.alarm(timeout_time)
try:
retval = f(*args)
except DecoratorTimeout:
return default
finally:
signal.signal(signal.SIGALRM, old_handler)
signal.alarm(0)
return retval
return f2
return timeout_function
  • 自完成groupby
def groupby(it, key):
"""
自完成groupby,itertool的groupby不能合并不接连可是相同的组, 且回来值是iter
:return: 字典目标
"""
groups = dict()
for item in it:
groups.setdefault(key(item), []).append(item)
return groups
  • cookie解析
def parse_cookie(string):
"""
解析cookie
:param string:
:return:
"""
results = re.findall('([^=]+)=([^\;]+);?\s?', string)
my_dict = {}
for item in results:
my_dict[item[0]] = item[1]
return my_dict
  • 查找字符串函数和类
def load_function(function_str):
"""
回来字符串表明的函数目标
:param function_str: module1.module2.function
:return: function
"""
mod_str, _sep, function_str = function_str.rpartition('.')
return getattr(load_module(mod_str), function_str)
load_class = load_function
  • 查找字符串模块
def load_module(module_str):
"""
回来字符串表明的模块
:param module_str: os.path
:return: os.path
"""
return __import__(module_str, fromlist=module_str.split(".")[-1])
  • 获取可用端口
def free_port():
"""
Determines a free port using sockets.
"""
free_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
free_socket.bind(('0.0.0.0', 0))
free_socket.listen(5)
port = free_socket.getsockname()[1]
free_socket.close()
return port
  • 线程安全装修器
def thread_safe(lock):
"""
对指定函数进行线程安全包装,需求供给锁
:param lock: 锁
:return:
"""
def decorate(func):
@wraps(func)
def wrapper(*args, **kwargs):
with lock:
return func(*args, **kwargs)
return wrapper
return decorate
  • 慢保存或提交
def call_later(callback, call_args=tuple(), immediately=True, interval=1):
"""
运用场景:
被装修的办法需求很多调用,随后需求调用保存办法,可是由于被装修的办法访问量很高,而保存办法开支很大
所以规划在装修办法继续调用必定距离后,再调用保存办法。规则距离内,不管调用多少次被装修办法,保存办法只会
调用一次,除非immediately=True
:param callback: 随后需求调用的办法名
:param call_args: 随后需求调用的办法所需求的参数
:param immediately: 是否当即调用
:param interval: 调用距离
:return:
"""
def decorate(func):
@wraps(func)
def wrapper(*args, **kwargs):
self = args[0]
try:
return func(*args, **kwargs)
finally:
if immediately:
getattr(self, callback)(*call_args)
else:
now = time.time()
if now - self.__dict__.get("last_call_time", 0) > interval:
getattr(self, callback)(*call_args)
self.__dict__["last_call_time"] = now
return wrapper
return decorate
  • 类中的线程安全装修器
def thread_safe_for_method_in_class(func):
"""
对类中的办法进行线程安全包装
:param func:
:return:
"""
@wraps(func)
def wrapper(*args, **kwargs):
self = args[0]
try:
self.lock.acquire()
return func(*args, **kwargs)
finally:
self.lock.release(濮)
return wrapper
  • 获取ip
def get_ip():
"""
获取局域网ip
:return:
"""
netcard_info = []
info = psutil.net_if_addrs()
for k,v in info.items():
for item in v:
if item[0] == 2 and not item[1]=='127.0.0.1':
netcard_info.append((k,item[1]))
if netcard_info:
return netcard_info[0][1]

下面供给一些基础设施

  • common_stop_start_control :供给开,关,重启,状况等命令行服务Singleton:单例元类Logger:供给日志服务SettingsWrapper: 供给装备信息服务ParallelMonitor: 运用Singleton构建, 多线程多进程一致管理器LoggingMonitor:内建Logger和SettingsService: 承继自ParallelMonitor,LoggingMonitor并完成common_stop_start_control 接口。编写微服务专用基类。ProxyPool:根据redis的署理池,承继自LoggerItemConsumer:kafka顾客,承继自ServiceItemProducer:kafka出产者,承继自ServiceRedisQueue:根据redis的行列FifoDiskQueue:耐久化 FIFO 行列Translate:翻译器,承继自ProxyPool,装置即可用的翻译器见https://github.com/ShichaoMa/translate_html