开发工具类/函数

此处提供了一些开发过程中较常用且不依赖于特殊场景和生产环境的类和方法。

将选取其中部分介绍用法。

auto_call

一个装饰器方法，在类中使用，提供成员方法名作为参数，该方法装饰过的成员方法会在被调用时先运行已提供方法名的方法。

典型应用场景

一个类方法使用时需要在另一个成员方法运行后再运行，则可将先运行的方法名传入 auto_call 后，装饰在该方法上。

例如指定initial方法为自动调用方法：

from deepfos.lib.utils import *

auto_initial = auto_call('initial')

编写类A，将 auto_initial 装饰在 say 方法上

class A:
    @staticmethod
    def initial():
        print('Initial A...')

    @classmethod
    @auto_initial
    def say(cls, words):
        print('Say:' + words)

调用时可见，initial 方法调用后，say 方法才被调用

A.say('Hi!')

Initial A...
Say:Hi!

ConcealableAttr

可隐藏变量，可读/写描述符，调用 expose 后会暴露变量，该变量可访问；调用 conceal 则会隐藏变量，此时访问会引发 AttributeError

典型应用场景

有一个属性值仅在特定上下文中有意义，为了防止在其他代码中意外访问到该变量而产生难以debug的错误，可以在进入上下文前将该变量暴露，退出上下文时将该变量隐藏。

定义类 T，其中的成员变量 attr 为可隐藏变量，默认值为 test

class T:
    attr = ConcealableAttr("test")

设置变量 t 为该类的对象，暴露其中的成员 attr

t = T()
T.attr.expose(t)

t.attr

'test'

隐藏 attr 后访问，则失败

T.attr.conceal(t)

t.attr

Attribute 'attr' is concealed.

get_ignore_case

类似于 dict.get，但忽略大小写，仅在字典很小时推荐使用。

t = {'Aa':123}

get_ignore_case(t, 'AA')

123

get_ignore_case(t, 'AB', 'not found')

'not found'

GroupDict

传入唯一键名集合 group，在新增字典的键名已存在在 group 中时，报错 KeyError

典型应用场景

多个字典在使用场景中有键名共享的关系。

设置变量 shared_key_group 为 Group 类的对象，此处 Group 类是用于维护共享键唯一性的集合

shared_key_group = Group()

设置变量 a, b 为 GroupDict类的对象，且键名共享 shared_key_group集合

a = GroupDict(shared_key_group)
b = GroupDict(shared_key_group)

设置 a中 A的值对应为 1，此时 a b的值

a['A'] = 1
a, b

({'A': 1}, {})

尝试设置 b的 A的值对应为 2

b['A'] = 2

'Key A has been existed in key_group. Cannot be added to current dict.'

设置失败，因为键 A已经存在于 a b所在的键集合中，而继续更新 a中的 A则是可以成功的

a['A'] = 'a'
a, b

({'A': 'a'}, {})

get_groupdicts

提供了对 GroupDict对象使用的进一步包装，获取 n 个字典，其中的键值共享一个集合，在字典间保持唯一

定义3个字典，共享同一个键集合

a, b, c = get_groupdicts(3)
a, b, c

({}, {}, {})

为 a 设置 A的值

a['A'] = 1
a, b, c

({'A': 1}, {}, {})

从 a 中取出 A

a.pop('A')

1

在 b和 c中分别设置 A的值

b['A'] = 2
c['A'] = 3

'Key A has been existed in key_group. Cannot be added to current dict.'

此时 a b c中的值

a, b, c

({}, {'A': 2}, {})

LazyList

元素延迟初始化的列表，元素的值只会在使用时被计算。

典型应用场景

计算结果不需要立即执行，而是可以在使用时执行，可以在高并发或异步的程序中节省时间开销。

编写一个计算总数的方法，指定 ll 为 LazyList 类的对象

def cal_sum(*args):
    print('calc')
    return sum(args)

ll = LazyList()

加入新元素为传入了参数 1,2,3 的 cal_sum方法的值

ll.append(cal_sum, 1, 2, 3)

可见，LazyList 在加入元素时未计算 cal_sum方法

print(ll[0])

calc
6

在被访问时， cal_sum方法才计算

print(ll[0])

6

再次访问则不会再调用cal_sum方法

LazyDict

元素延迟初始化的字典，键的值只会在使用时被计算。

典型应用场景

计算结果不需要立即执行，而是可以在使用时执行，可以在高并发或异步的程序中节省时间开销。

编写一个计算总数的方法，指定 ll 为 LazyDict 类的对象

def cal_sum(*args):
    print('calc')
    return sum(args)

ld = LazyDict()

设置键 a的值为传入了参数 1,2,3 的 cal_sum方法的值

ld['a'] = (cal_sum, 1, 2, 3)

可见，LazyDict 在设置时未计算 cal_sum方法

print(ld['a'])

calc
6

在被访问时， cal_sum方法才计算

print(ld['a'])

6

再次访问则不会再调用cal_sum方法

retry

一个装饰器方法，在被装饰函数执行失败时，重新执行。

编写一个必定运行失败的方法，用 retry 装饰，每隔2秒重试1次，共重试3次

@retry(retries=3,wait=2)
def always_fail():
    raise ValueError('Failure occurs!')

always_fail()

2022-02-08 12:04:18.015 | 17972 | MainThread | ERROR    | deepfos.lib.utils:record_retry:269 - Func: 'always_fail' failed. Start 1 times retry in 2 secs.
Traceback (most recent call last):

  ......, in always_fail
    raise ValueError('Failure occurs!')

ValueError: Failure occurs!
2022-02-08 12:04:20.034 | 17972 | MainThread | ERROR    | deepfos.lib.utils:record_retry:269 - Func: 'always_fail' failed. Start 2 times retry in 2 secs.
Traceback (most recent call last):

  ......, in always_fail
    raise ValueError('Failure occurs!')

ValueError: Failure occurs!
2022-02-08 12:04:22.040 | 17972 | MainThread | ERROR    | deepfos.lib.utils:record_retry:269 - Func: 'always_fail' failed. Start 3 times retry in 2 secs.
Traceback (most recent call last):

  ......, in always_fail
    raise ValueError('Failure occurs!')

ValueError: Failure occurs!

可指定 fix 方法，在重试前执行。

def trying_to_fix():
    print('Rollback...')

@retry(retries=3,wait=2,fix=trying_to_fix)
def always_fail():
    raise ValueError('Failure occurs!')

always_fail()

2022-02-08 12:04:24.078 | 17972 | MainThread | ERROR    | deepfos.lib.utils:record_retry:269 - Func: 'always_fail' failed. Start 1 times retry in 2 secs.
Traceback (most recent call last):

  ......, in always_fail
    raise ValueError('Failure occurs!')

ValueError: Failure occurs!
Rollback...
2022-02-08 12:04:26.089 | 17972 | MainThread | ERROR    | deepfos.lib.utils:record_retry:269 - Func: 'always_fail' failed. Start 2 times retry in 2 secs.
Traceback (most recent call last):

  ......, in always_fail
    raise ValueError('Failure occurs!')

ValueError: Failure occurs!
Rollback...
2022-02-08 12:04:28.099 | 17972 | MainThread | ERROR    | deepfos.lib.utils:record_retry:269 - Func: 'always_fail' failed. Start 3 times retry in 2 secs.
Traceback (most recent call last):

  ......, in always_fail
    raise ValueError('Failure occurs!')

ValueError: Failure occurs!