in source/setup/lambda/crhelper/resource_helper.py [0:0]
def __init__(self, json_logging=False, log_level='DEBUG', boto_level='ERROR', polling_interval=2, sleep_on_delete=120, ssl_verify=None):
self._sleep_on_delete = sleep_on_delete
self._create_func = None
self._update_func = None
self._delete_func = None
self._poll_create_func = None
self._poll_update_func = None
self._poll_delete_func = None
self._timer = None
self._init_failed = None
self._json_logging = json_logging
self._log_level = log_level
self._boto_level = boto_level
self._send_response = False
self._polling_interval = polling_interval
self.Status = ""
self.Reason = ""
self.PhysicalResourceId = ""
self.StackId = ""
self.RequestId = ""
self.LogicalResourceId = ""
self.Data = {}
self.NoEcho = False
self._event = {}
self._context = None
self._response_url = ""
self._sam_local = os.getenv('AWS_SAM_LOCAL')
self._region = os.getenv('AWS_REGION')
self._ssl_verify = ssl_verify
try:
if not self._sam_local:
self._lambda_client = boto3.client('lambda', region_name=self._region, verify=self._ssl_verify)
self._events_client = boto3.client('events', region_name=self._region, verify=self._ssl_verify)
self._logs_client = boto3.client('logs', region_name=self._region, verify=self._ssl_verify)
if json_logging:
log_helper.setup(log_level, boto_level=boto_level, RequestType='ContainerInit')
else:
log_helper.setup(log_level, formatter_cls=None, boto_level=boto_level)
except Exception as e:
logger.error(e, exc_info=True)
self.init_failure(e)