fromsimple_cacheimportbuild_cacheREDIS_CONFIG={"host": "host of redis clusters", "port": 6379, "password": "password of redis clusters" } algo_model_cache=build_cache(db=99, config_json=REDIS_CONFIG) if__name__=='__main__': fromhashlibimportmd5importjsondefcontent_md5(data): """计算data的MD5值，经过Base64编码并返回str类型。 返回值可以直接作为HTTP Content-Type头部的值 """ifisinstance(data, str): data=data.encode(encoding='utf-8') m=md5(data) returnm.hexdigest().upper() defkey_calculate(model, data, namespace='production', **kwargs): data=json.dumps(data) returnmodel+"_"+namespace+"_"+content_md5(data) defcondition(model_result, model, data, namespace='production', **kwargs): ifmodel_result.get('status', '') =='SUCCESS': returnTruereturnFalse@algo_model_cache(cache_name="test", cache_mode='cache', expire=60, key_func=key_calculate, cache_condition_func=condition)defcache_function(cached_func, data, namespace='production', **kwargs): """算法"""model_result=cached_func(data, namespace) returnmodel_resultdeffunc(data, namespace): data["result"] ="model_result"data["namespace"] ="namespace"returndatainput_data={"task_id": "test_cache", "data":{"curtain_size": [ 1280, 720 ] } } result=cache_function(func, data=input_data, namespace='dev') print(result)