py-otils

python

Posted by jiang on July 23, 2020

py-otils

说明

  • python office utils for human beings
  • 常用工具包装模块

安装

1
python setup.py install

示例

线程分发

1
2
3
4
5
from otils import Manager

tm = Manager(lambda x: x + 10, list(range(10)), num=2, test=True)
tm.do_work()

进程分发

  • windows下进程执行入口必须为if __name__ == '__main__':
  • handler function 不支持lambda
1
2
3
4
5
6
7
8
9
10
11
12
from otils import Manager

def foo(x):
    return x + 10

def Process_test():
    tm = Manager(foo, list(range(10)), num=2, wtype='process')
    tm.do_work()

if __name__ == '__main__':
    Process_test()

协程分发

1
2
3
4
5
6
def Coro_test():
    add_100 = wadd(100)
    add_10 = wadd(10)
    func_list = [add_10, add_100]
    tm = Manager(func_list, list(range(11)), num=3, test=True, wtype='coro',sleep=2,timeout=10)
    tm.do_work()

处理函数

  • 处理函数支持单函数传递,也支持函数列表
  • 对于需要特殊参数的函数可以通过闭包或者偏函数来解决;将参数封装到item内,由处理函数解包处理也是一种办法

  • 偏函数
1
2
3
4
5
6
7
8
9
10
11
12
13
from functools import partial

def add(m, n):
    return m+n

def Thread_partial_test():
    add_100 = partial(add,100)
    add_10 = partial(add, 10)

    func_list = [add_10, add_100]
    tm = Manager(func_list, list(range(11)), num=2, test=True)
    tm.do_work()

  • 闭包
  • windows下多进程需要序列化,闭包不支持序列化,所以windows多进程闭包不支持
1
2
3
4
5
6
7
8
9
10
11
12
13
def wadd(x):
    def add(m):
        return m+x
    return add

def Thread_closure_test():
    add_100 = wadd(100)
    add_10 = wadd(10)

    func_list = [add_10, add_100]
    tm = Manager(func_list, list(range(11)), num=2, test=True)
    tm.do_work()

其他工具

  • 文本与xlsx读写
1
2
3
4
5
6
7
8
9
10
11
from otils import Writer,Reader
wr = Writer('id',suffix='txt',delimiter=',')
data = [['1','2','3'],['4','5','6']]
wr.write(data)

wr = Writer('id',suffix='xlsx')
data = [['1','2','3'],['4','5','6']]
wr.write(data)

rd = Reader('id',suffix='txt',delimiter=',')
rd.read()
  • 时间函数
1
2
3
4
from otils import now_add,now

now_add(7)
now()
  • 文件大小
1
2
3
from otils import get_size
get_size(path)

  • 批量执行模块内无参数方法
1
2
3
4
5
from otils import batch_run_func

module = __import__('t_manager')
batch_run_func(module)

  • 匹配字符之间的子串
1
2
from otils import get_between
get_between('abchdhshedmm','ch','ed')
  • 函数计时装饰器
1
2
3
4
5
6
7
8
9
 
@cost()
def Coro_test(timeout=10):
    add_100 = wadd(100)
    add_10 = wadd(10)
    func_list = [add_10, add_100]
    tm = Manager(func_list, list(range(11)), num=3, test=True, wtype='coro',sleep=2,timeout=timeout)
    tm.do_work()