import sys import enum import os import logging import traceback #from paddle_serving_server.pipeline import ResponseOp import threading import inspect import traceback import functools import re from .proto import pipeline_service_pb2_grpc, pipeline_service_pb2 from .util import ThreadIdGenerator from paddle_serving_server.util import kill_stop_process_by_pid _LOGGER = logging.getLogger(__name__) class CustomExceptionCode(enum.Enum): """ Add new Exception 0 Success 50 ~ 99 Product error 3000 ~ 3999 Internal service error 4000 ~ 4999 Conf error 5000 ~ 5999 User input error 6000 ~ 6999 Timeout error 7000 ~ 7999 Type Check error 8000 ~ 8999 Internal communication error 9000 ~ 9999 Inference error 10000 Other error """ OK = 0 PRODUCT_ERROR = 50 NOT_IMPLEMENTED = 3000 CLOSED_ERROR = 3001 NO_SERVICE = 3002 INIT_ERROR = 3003 CONF_ERROR = 4000 INPUT_PARAMS_ERROR = 5000 TIMEOUT = 6000 TYPE_ERROR = 7000 RPC_PACKAGE_ERROR = 8000 CLIENT_ERROR = 9000 UNKNOW = 10000 class ProductErrCode(enum.Enum): """ ProductErrCode is to record business error codes. the ProductErrCode number ranges from 51 to 99 product developers can directly add error code into this class. """ pass class CustomException(Exception): """ An self-defined exception class Usage : raise CustomException(CustomExceptionCode.exceptionCode, errorMsg, isSendToUser=False) Args : exceptionCode : CustomExceptionCode or ProductErrCode errorMsg : string message you want to describe the error isSendToUser : whether send to user or just record in errorlog Return : An string of error_info """ def __init__(self, exceptionCode, errorMsg, isSendToUser=False): super().__init__(self) self.error_info = "\n\texception_code: {}\n"\ "\texception_type: {}\n"\ "\terror_msg: {}\n"\ "\tis_send_to_user: {}".format(exceptionCode.value, CustomExceptionCode(exceptionCode).name, errorMsg, isSendToUser) def __str__(self): return self.error_info class ErrorCatch(): """ An decorator class to catch error for method or function. Usage : @ErrorCatch Args : None Returns: tuple(res, response) res is the original funciton return response includes erro_no and erro_msg """ def __call__(self, func): if inspect.isfunction(func) or inspect.ismethod(func): @functools.wraps(func) def wrapper(*args, **kw): try: res = func(*args, **kw) except CustomException as e: if "log_id" in kw.keys(): log_id = kw["log_id"] elif "logid_dict" in kw.keys() and "data_id" in kw.keys(): log_id = kw["logid_dict"].get(kw["data_id"]) else: log_id = 0 resp = pipeline_service_pb2.Response() _LOGGER.error("\nLog_id: {}\n{}Classname: {}\nFunctionName: {}\nArgs: {}".format(log_id, traceback.format_exc(), func.__qualname__, func.__name__, args)) split_list = re.split("\n|\t|:", str(e)) resp.err_no = int(split_list[3]) resp.err_msg = "Log_id: {} Raise_msg: {} ClassName: {} FunctionName: {}".format(log_id, split_list[9], func.__qualname__ ,func.__name__ ) is_send_to_user = split_list[-1].replace(" ", "") if is_send_to_user == "True": return (None, resp) else: print("Erro_Num: {} {}".format(resp.err_no, resp.err_msg)) print("Init error occurs. For detailed information. Please look up pipeline.log.wf in PipelineServingLogs by log_id.") kill_stop_process_by_pid("kill", os.getpgid(os.getpid())) except Exception as e: if "log_id" in kw.keys(): log_id = kw["log_id"] elif "logid_dict" in kw.keys() and "data_id" in kw.keys(): log_id = kw["logid_dict"].get(kw["data_id"]) else: log_id = 0 resp = pipeline_service_pb2.Response() _LOGGER.error("\nLog_id: {}\n{}Classname: {}\nFunctionName: {}".format(log_id, traceback.format_exc(), func.__qualname__, func.__name__)) resp.err_no = CustomExceptionCode.UNKNOW.value resp.err_msg = "Log_id: {} Raise_msg: {} ClassName: {} FunctionName: {}".format(log_id, str(e).replace("\'", ""), func.__qualname__ ,func.__name__ ) return (None, resp) else: resp = pipeline_service_pb2.Response() resp.err_no = CustomExceptionCode.OK.value resp.err_msg = "" return (res, resp) return wrapper def ParamChecker(function): @functools.wraps(function) def wrapper(*args, **kwargs): # fetch the argument name list. parameters = inspect.signature(function).parameters argument_list = list(parameters.keys()) # fetch the argument checker list. checker_list = [parameters[argument].annotation for argument in argument_list] # fetch the value list. value_list = [inspect.getcallargs(function, *args, **kwargs)[argument] for argument in inspect.getfullargspec(function).args] # initialize the result dictionary, where key is argument, value is the checker result. result_dictionary = dict() for argument, value, checker in zip(argument_list, value_list, checker_list): result_dictionary[argument] = check(argument, value, checker, function) # fetch the invalid argument list. invalid_argument_list = [key for key in argument_list if not result_dictionary[key]] # if there are invalid arguments, raise the error. if len(invalid_argument_list) > 0: raise CustomException(CustomExceptionCode.INPUT_PARAMS_ERROR, "invalid arg list: {}".format(invalid_argument_list)) # check the result. result = function(*args, **kwargs) checker = inspect.signature(function).return_annotation if not check('return', result, checker, function): raise CustomException(CustomExceptionCode.INPUT_PARAMS_ERROR, "invalid return type") # return the result. return result return wrapper def check(name, value, checker, function): if isinstance(checker, (tuple, list, set)): return True in [check(name, value, sub_checker, function) for sub_checker in checker] elif checker is inspect._empty: return True elif checker is None: return value is None elif isinstance(checker, type): return isinstance(value, checker) elif callable(checker): result = checker(value) return result class ParamVerify(object): @staticmethod def int_check(c, lower_bound=None, upper_bound=None): if not isinstance(c, int): return False if isinstance(lower_bound, int) and isinstance(upper_bound, int): return c >= lower_bound and c <= upper_bound return True @staticmethod def file_check(f): if not isinstance(f, str): return False if os.path.exist(f): return True else: return False @staticmethod def check_feed_dict(feed_dict, feed_list): if not isinstance(feed_dict, dict): return False # read model config, try catch and feed_dict_key_size = len(feed_dict.keys()) if len(feed_dict.keys()) != len(feed_list): return False for key in feed_list: if key in feed_dict.keys(): return False return True ErrorCatch = ErrorCatch()