forked from HDE/python-lambda-local
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
143 lines (106 loc) · 3.73 KB
/
main.py
File metadata and controls
143 lines (106 loc) · 3.73 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
'''
Copyright 2015-2018 HDE, Inc.
Licensed under MIT.
'''
import imp
import sys
import traceback
import json
import logging
import os
import timeit
from botocore.vendored.requests.packages import urllib3
import multiprocessing
from . import event
from . import context
from .environment_variables import set_environment_variables, export_variables
from .timeout import time_limit
from .timeout import TimeoutException
logging.basicConfig(stream=sys.stdout,
level=logging.INFO,
format='[%(name)s - %(levelname)s - %(asctime)s] %(message)s')
urllib3.disable_warnings()
ERR_TYPE_EXCEPTION = 0
ERR_TYPE_TIMEOUT = 1
EXITCODE_ERR = 1
class ContextFilter(logging.Filter):
def __init__(self, context):
super(ContextFilter, self).__init__()
self.context = context
def filter(self, record):
record.aws_request_id = self.context.aws_request_id
return True
def call(func, event, context, environment_variables={}):
export_variables(environment_variables)
return _runner(func, event, context)
def run(args):
# set env vars if path to json file was given
set_environment_variables(args.environment_variables)
e = event.read_event(args.event)
c = context.Context(
args.timeout,
invoked_function_arn=args.arn_string,
function_version=args.version_name)
if args.library is not None:
load_lib(args.library)
func = load(c.aws_request_id, args.file, args.function)
(result, err_type) = _runner(func, e, c)
if err_type is not None:
sys.exit(EXITCODE_ERR)
def _runner(func, event, context):
logger = logging.getLogger()
logger.info("Event: {}".format(event))
logger.info("START RequestId: {} Version: {}".format(
context.aws_request_id, context.function_version))
queue = multiprocessing.Queue()
p = multiprocessing.Process(
target=execute_in_process,
args=(queue, func, event, context,))
p.start()
(result, err_type, duration) = queue.get()
p.join()
logger.info("END RequestId: {}".format(context.aws_request_id))
duration = "{0:.2f} ms".format(duration)
logger.info("REPORT RequestId: {}\tDuration: {}".format(
context.aws_request_id, duration))
if type(result) is TimeoutException:
logger.error("RESULT:\n{}".format(result))
else:
logger.info("RESULT:\n{}".format(result))
return (result, err_type)
def load_lib(path):
sys.path.append(os.path.abspath(path))
def load(request_id, path, function_name):
mod_name = 'request-' + str(request_id)
file_path = os.path.abspath(path)
file_directory = os.path.dirname(file_path)
sys.path.append(file_directory)
mod = imp.load_source(mod_name, path)
func = getattr(mod, function_name)
return func
def execute(func, event, context):
err_type = None
logger = logging.getLogger()
log_filter = ContextFilter(context)
logger.addFilter(log_filter)
try:
with time_limit(context._timeout_in_seconds):
result = func(event, context._activate())
except TimeoutException as err:
result = err
err_type = ERR_TYPE_TIMEOUT
except:
err = sys.exc_info()
result = json.dumps({
"errorMessage": str(err[1]),
"stackTrace": traceback.extract_tb(err[2]),
"errorType": err[0].__name__
}, indent=4, separators=(',', ': '))
err_type = ERR_TYPE_EXCEPTION
return result, err_type
def execute_in_process(queue, func, event, context):
start_time = timeit.default_timer()
result, err_type = execute(func, event, context)
end_time = timeit.default_timer()
duration = (end_time - start_time) * 1000
queue.put((result, err_type, duration))