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
91 lines (70 loc) · 2.26 KB
/
main.py
File metadata and controls
91 lines (70 loc) · 2.26 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
'''
Copyright 2015 HDE, Inc.
Licensed under MIT.
'''
import imp
import sys
import traceback
import json
import logging
import uuid
import os
import timeit
from botocore.vendored.requests.packages import urllib3
import event
import context
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()
def run(args):
e = event.read_event(args.event)
c = context.Context(args.timeout, args.arn_string, args.version_name)
if args.library is not None:
load_lib(args.library)
request_id = uuid.uuid4()
func = load(request_id, args.file, args.function)
logger = logging.getLogger()
result = None
logger.info("Event: {}".format(e))
logger.info("START RequestId: {}".format(request_id))
start_time = timeit.default_timer()
try:
result = execute(func, e, c)
except TimeoutException as te:
result = te
end_time = timeit.default_timer()
logger.info("END RequestId: {}".format(request_id))
if type(result) is TimeoutException:
logger.error("RESULT:\n{}".format(result))
else:
logger.info("RESULT:\n{}".format(result))
duration = "{0:.2f} ms".format((end_time - start_time) * 1000)
logger.info("REPORT RequestId: {}\tDuration: {}".format(
request_id, duration))
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):
try:
with time_limit(context.timeout):
result = func(event, context.activate())
except TimeoutException as err:
raise err
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=(',', ': '))
return result