使用(Usage)

装饰器

decorators.common

handle_exception(func=None, default=False, is_raise=False, retry_for=Exception, max_retries=1, retry_delay=0, retry_jitter=True, log_args=True, logger_name='pykit_tools.error', logger_level=logging.ERROR, logger_pre_level=logging.WARNING)

装饰器 用于捕获函数异常，并在出现异常的时候返回默认值

Parameters:
  • func (Optional[Callable], default: None ) –

    函数

  • default (Any, default: False ) –

    出现异常后的默认值

  • is_raise (bool, default: False ) –

    是否抛出异常；设置True时，default参数无效且一定会抛出异常，主要用于重试场景最后依然抛出异常

  • retry_for (Union[Type, Tuple], default: Exception ) –

    需要重试的异常类/异常元组，仅当异常匹配才进行重试

  • max_retries (int, default: 1 ) –

    最大重试次数

  • retry_delay (int, default: 0 ) –

    重试等待时间，默认值0（不推荐开启）

  • retry_jitter (bool, default: True ) –

    重试抖动，用于将随机性引入指数退避延迟，以防止队列中的所有任务同时执行； 若设置为true, 随机范围值在[0, retry_delay]之间，随机值为真实delay时间

  • log_args (bool, default: True ) –

    异常时将参数输出到日志

  • logger_name (str, default: 'pykit_tools.error' ) –

    日志名称，仅记录异常时使用

  • logger_level (int, default: ERROR ) –

    异常时设置日志的级别

  • logger_pre_level (int, default: WARNING ) –

    重试最后一次之前的日志级别，避免多次重试会有多次错误输出
Returns:
  • Callable –

    function

time_record(func=None, format_key=None, format_ret=None, logger_name='pykit_tools.error', logger_level=logging.ERROR)

装饰器 函数耗时统计

Parameters:
  • func (Optional[Callable], default: None ) –
  • format_key (Optional[Callable], default: None ) –

    根据函数输入的参数，格式化日志记录的唯一标记key

  • format_ret (Optional[Callable], default: None ) –

    根据函数返回的结果，格式化日志记录的结果ret

  • logger_name (str, default: 'pykit_tools.error' ) –

    日志名称，仅记录异常时使用

  • logger_level (int, default: ERROR ) –

    异常时设置日志的级别
Returns:
  • Callable –

    function

decorators.cache

method_deco_cache(func=None, key=None, timeout=60, scene=CacheScene.DEFAULT.value, cannot_cache=(None, False), cache_client=None, cache_max_length=33554432, logger_name='pykit_tools.error', logger_level=logging.ERROR)

装饰器 方法缓存结果, 只能缓存json序列化的数据类型

注意：若是在类的实例方法上使用，需要注意 self 参数的影响，可设置key或者将类单例后使用

Parameters:
  • func (Optional[Callable], default: None ) –

    可以在放在参数添加 scene=CacheScene.DEGRADED.value,可以强制进行刷新

  • key (Optional[Union[str, Callable]], default: None ) –

    str, 缓存数据存储的key； 也可以传递func，根据参数动态构造

  • timeout (int, default: 60 ) –

    缓存超时时间，单位 秒(s)

  • scene (str, default: value ) –

    默认使用场景 CacheScene

  • cannot_cache (Union[List, Tuple], default: (None, False) ) –

    元组，不允许缓存的数值 传递False或者None表示缓存所有类型的结果数据，若是仅None不缓存一定要设置值为元组(None, ) 若是设置的函数，则根据返回数据作为输入参数、输出bool表示不允许缓存；

  • cache_client (Any, default: None ) –

    缓存client对象

  • cache_max_length (int, default: 33554432 ) –

    序列化后缓存的字符串最大长度限制， 此处设置最大缓存 32M = 32 * 1024 * 1024 若是redis, A String value can be at max 512 Megabytes in length.

  • logger_name (str, default: 'pykit_tools.error' ) –

    日志名称

  • logger_level (int, default: ERROR ) –

    异常时设置日志的级别

Returns: function

singleton_refresh_regular(cls=None, timeout=5)

装饰器 带定时刷新的单例装饰器

应用场景：例如某对象实例化后带有session相关信息，有一定有效期的情况可以在类上加上该装饰器

Parameters:
  • cls (Optional[Type], default: None ) –

    类

  • timeout (int, default: 5 ) –

    单例使用超时时间，单位秒(s)
Returns:
  • Callable –

    function

示例：

@singleton_refresh_regular
class YouClass(Singleton):
    pass

decorators.req_utils

requests_logger(func=None, default_ua='', logger_name='pykit_tools.requests', logger_level=logging.ERROR)

装饰器 应用于对 requests 库的请求进行日志记录.

扩展 requests 请求函数可传递的参数：

  • log_request: 是否记录请求的参数，默认True
  • log_response: 是否记录响应内容，默认True，设置为False后可通过参数legal_codes控制记录响应内容
  • legal_codes: 合法的响应状态码列表/元组，仅当响应状态码不在列表/元组中才记录响应内容，默认None记录所有响应内容
  • format_resp: 格式化响应内容的函数，用于格式化响应内容，默认None使用response.text
  • raise_for: 需要抛出的异常类型/异常元组，仅当异常匹配才抛出异常，默认None不抛出异常
Parameters:
  • func (Optional[Callable], default: None ) –
  • default_ua (str, default: '' ) –

    默认的 User-Agent

  • logger_name (str, default: 'pykit_tools.requests' ) –

    日志名称，仅记录异常时使用

  • logger_level (int, default: ERROR ) –

    异常时设置日志的级别
Returns:
  • function( Callable ) –

示例：

import typing
import requests
from pykit_tools.decorators.req_utils import requests_logger

class BaseRequest(object):

    def __init__(self, default_ua: str = "", logger_name: str = "pykit_tools.requests"):
        self.request = requests_logger(requests.request, default_ua=default_ua, logger_name=logger_name)

    def get(self, url: str, **kwargs: typing.Any) -> requests.Response:
        return self.request("GET", url, **kwargs)

    def post(self, url: str, **kwargs: typing.Any) -> requests.Response:
        return self.request("POST", url, **kwargs)

    def put(self, url: str, **kwargs: typing.Any) -> requests.Response:
        return self.request("PUT", url, **kwargs)

    def delete(self, url: str, **kwargs: typing.Any) -> requests.Response:
        return self.request("DELETE", url, **kwargs)

    def patch(self, url: str, **kwargs: typing.Any) -> requests.Response:
        return self.request("PATCH", url, **kwargs)

    def head(self, url: str, **kwargs: typing.Any) -> requests.Response:
        return self.request("HEAD", url, **kwargs)

    def options(self, url: str, **kwargs: typing.Any) -> requests.Response:
        return self.request("OPTIONS", url, **kwargs)

br = BaseRequest()
# 控制日志输出内容
br.get("https://www.baidu.com", log_request=False, log_response=True)
# 控制异常抛出
br.get("https://www.baidu.com", raise_for=Exception)

装饰器相关

decorators.cache.CacheScene

Bases: ChoiceEnum

枚举 缓存场景类型，定义值详见源码。

应用于装饰器 method_deco_cache

Source code in pykit_tools/decorators/cache.py 
43
44
45
46
47
48
49
50
51
52
class CacheScene(ChoiceEnum):
    """
    `枚举` 缓存场景类型，定义值详见源码。

    应用于装饰器 [method_deco_cache](./#decorators.cache.method_deco_cache)
    """

    DEFAULT = ("default", "优先使用缓存，无缓存执行函数")
    DEGRADED = ("degraded", "优先执行函数，失败后降级使用缓存")  # 执行函数成功后，会更新缓存数据
    SKIP = ("skip", "忽略使用缓存，直接执行函数")  # 执行函数成功后，会更新缓存数据

decorators.cache

set_global_cache_client(client)

设置全局缓存client对象，用于存储缓存数据

Parameters:
  • client (Any) –
注意

client必须有get(key)和set(key, value, timeout)方法，一般使用redis客户端

get_global_cache_client()

获取全局缓存client对象

Returns:
  • Any –

    cache client

日志相关

log.adapter

LoggerFormatAdapter

Bases: LoggerAdapter

日志按照字段格式化输出

示例 
# 其中 get_format_logger 中使用该类
timer_common_logger = get_format_logger("pykit_tools.timer", ["location", "key", "cost", "ret"])

timer_common_logger.info(dict(location="my-method", key="my-key", cost=3, ret=True))
# 日志输出>> "my-method my-key 3 True"
__init__(logger, extra, fields=None, delimiter=' ', fmt='{}')

初始化构造对象

Parameters:
  • logger (Logger) –

    通过 logging.getLogger("name") 获得

  • extra (dict) –

    额外字段

  • fields (Optional[Union[List[str], Tuple[str]]], default: None ) –

    自定义的字段

  • delimiter (str, default: ' ' ) –

    各字段间分隔符，默认空格

  • fmt (str, default: '{}' ) –

    自定义格式化字符串，用于最后 fmt.format(dict_msg)
process(msg, kwargs)

覆写 logging.LoggerAdapter.process 处理自定义格式化日志信息

Parameters:
  • msg (Dict) –

    使用日志输出时使用字典结构

  • kwargs (Any) –

    logger输出定义其他参数
Returns:
  • str –

    msg 日志内容字符串

  • dict –

    kwargs 额外参数

get_format_logger(name, fields, delimiter=' ', extra=None)

根据 LoggerFormatAdapter 获取 logger

Parameters:
  • name (str) –

    名称

  • fields (Union[List, Tuple]) –

    自定义字段

  • delimiter (str, default: ' ' ) –

    字段间分隔符

  • extra (Optional[Dict], default: None ) –

    额外参数
Returns:

log.handlers

MultiProcessTimedRotatingFileHandler

Bases: TimedRotatingFileHandler

Similar with logging.TimedRotatingFileHandler, while this one is Multi process safe.

多进程使用的LoggerHandler

设计模式

patterns.singleton

Singleton

可以提供给类直接继承

SingletonMeta

Bases: type

设计模式：单例类

示例 
class YouClass(metaclass=SingletonMeta)
    pass

其他

cmd

exec_command(command, timeout=60, log_cmd=False, err_max_length=1024, logger_name='pykit_tools.cmd', logger_level=logging.ERROR, popen_kwargs=None)

执行shell命令

Parameters:
  • command (str) –

    要执行的命令

  • timeout (int, default: 60 ) –

    超时时间，单位秒(s)

  • log_cmd (bool, default: False ) –

    是否记录命令日志，默认不记录（仅在异常时记录异常日志）

  • err_max_length (int, default: 1024 ) –

    错误输出最大长度，超过该长度则截断; 0表示不截断

  • logger_name (str, default: 'pykit_tools.cmd' ) –

    日志名称

  • logger_level (int, default: ERROR ) –

    异常时设置日志的级别

  • popen_kwargs (Optional[Dict], default: None ) –

    透传 subprocess.Popen 的参数
Returns:
  • int –

    code 系统执行返回，等于0表示成功

  • str –

    stdout 执行输出

  • str –

    stderr 执行错误输出

str_tool

base64url_decode(value)

对URL安全编码进行解码

Parameters:
  • value (str) –

    输入编码字符串
Returns:
  • str –

    解码后字符串

base64url_encode(value)

对内容进行URL安全的Base64编码，需要将结果中的部分编码替换：

  • 将结果中的加号 + 替换成短划线 -;
  • 将结果中的正斜线 / 替换成下划线 _;
  • 将结果中尾部的所有等号 = 省略。
Parameters:
  • value (str) –

    输入字符串
Returns:
  • str –

    返回编码后字符串

compute_md5(*args, **kwargs)

根据输入的参数计算出唯一值（将参数值拼接后最后计算md5）

Parameters:
  • *args (Any, default: () ) –

    输入的参数

  • **kwargs (Any, default: {} ) –

    输入的k-v参数
Returns:
  • str –

    唯一值

is_number(s)

判断字符串是否是数值

Parameters:
  • s (str) –

    输入字符串
Returns:
  • bool –

    返回判断结果

str_to_number(s)

字符串转化成数字

Parameters:
  • s (str) –

    输入字符串
Returns:
  • Union[int, float] –

    输出数值，若是不能转化会抛出异常

utils

CacheMap

Bases: object

缓存对象

注意

若是key太多，容易OOM内存溢出； 且进程销毁会回收

clean()

清理过期的数据

clear()

清理所有缓存过的数据

delete(key)

根据key删除数据

Parameters:
  • key (str) –
Returns:
  • Any –

    返回删除的值
get(key)

根据key获取缓存的数据

Parameters:
  • key (str) –
Returns:
  • Any –

    数据值
set(key, value, timeout=60)

根据key设置数据值value

Parameters:
  • key (str) –

    键

  • value (Any) –

    值

  • timeout (int, default: 60 ) –

    超时时间，单位秒(s)
Returns:
  • Any –

    值

find_method_by_str(method_path)

通过字符串初始化方法

Parameters:
  • method_path (str) –

    方法路径, eg: "pykit_tools.utils.get_caller_location"
Returns:
  • Optional[Callable] –

    function，未找到返回None

get_caller_location(caller)

或者 类/方法 在项目中的路径 Args: caller: 方法或者类

Returns:
  • str –

    路径字符串