"""faas handler"""
import logging
from typing import List
from yr.common.utils import err_to_str
from yr.err_type import ErrorCode, ErrorInfo, ModuleCode
from yr.executor.faas_executor import faas_shutdown_handler
from yr.executor.handler_intf import HandlerIntf
_logger = logging.getLogger(__name__)
class FaasHandler(HandlerIntf):
"""
FaaS handler
"""
def __init__(self):
pass
def execute_function(self, func_meta, args: List, invoke_type, return_num: int, is_actor_async: bool):
"""execute function"""
pass
def shutdown(self, grace_period_second: int) -> ErrorInfo:
"""shutdown"""
try:
return faas_shutdown_handler(grace_period_second)
except Exception as e:
_logger.exception(e)
return ErrorInfo(ErrorCode.ERR_INNER_SYSTEM_ERROR, ModuleCode.RUNTIME, err_to_str(e))
def before_snapshot(self) -> ErrorInfo:
"""faas has no snapshot hook by default"""
return ErrorInfo()
def after_snapstart(self) -> ErrorInfo:
"""faas has no snapstart hook by default"""
return ErrorInfo()