demo2.py

def cache_wrap(func: typing.Callable) -> typing.Callable:
    @functools.wraps(func)
    async def func_wrap(*args, **kwargs) -> typing.Dict[str, typing.Any]:
        """

        :param phone:
        :param kwargs:
        :return: 所有有标签的用户列表
        """
        need_cache = kwargs.get("need_cache")
        signature = generate_signature(*args, **kwargs)
        # if idcard is None or name is None or phone is None:
        #     return {}
        temp = get_cache_data(signature, type(args[0]).__name__)
        temp_result = {}
        if temp is None and need_cache:
            temp_result = {}
        elif temp is None and not need_cache:
            temp_result = await func(*args, **kwargs)
            cache_data(signature, type(args[0]).__name__, temp_result)
        elif temp is not None:
            temp_result = temp.get("data")
            args = list(args)
            kwargs.pop("session")
            LOGGER.info("elk data{}" + type(args.pop(0)).__name__ + "," +
                        json.dumps(args) + "，" + json.dumps(kwargs))

        return temp_result

    return func_wrap