/
invoke.py
105 lines (81 loc) · 2.99 KB
/
invoke.py
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
# copy/pasted entirely as is from:
# https://github.com/serverless/serverless/blob/v1.50.0/lib/plugins/aws/invokeLocal/invoke.py
import subprocess
import argparse
import json
import logging
import sys
from time import strftime, time
from importlib import import_module
class FakeLambdaContext(object):
def __init__(self, name='Fake', version='LATEST', timeout=6, **kwargs):
self.name = name
self.version = version
self.created = time()
self.timeout = timeout
for key, value in kwargs.items():
setattr(self, key, value)
def get_remaining_time_in_millis(self):
return int(max((self.timeout * 1000) - (int(round(time() * 1000)) - int(round(self.created * 1000))), 0))
@property
def function_name(self):
return self.name
@property
def function_version(self):
return self.version
@property
def invoked_function_arn(self):
return 'arn:aws:lambda:serverless:' + self.name
@property
def memory_limit_in_mb(self):
return '1024'
@property
def aws_request_id(self):
return '1234567890'
@property
def log_group_name(self):
return '/aws/lambda/' + self.name
@property
def log_stream_name(self):
return strftime('%Y/%m/%d') +'/[$' + self.version + ']58419525dade4d17a495dceeeed44708'
@property
def log(self):
return sys.stdout.write
logging.basicConfig()
parser = argparse.ArgumentParser(
prog='invoke',
description='Runs a Lambda entry point (handler) with an optional event',
)
parser.add_argument('handler_path',
help=('Path to the module containing the handler function,'
' omitting ".py". IE: "path/to/module"'))
parser.add_argument('handler_name', help='Name of the handler function')
if __name__ == '__main__':
args = parser.parse_args()
# this is needed because you need to import from where you've executed sls
sys.path.append('.')
module = import_module(args.handler_path.replace('/', '.'))
handler = getattr(module, args.handler_name)
# Keep a reference to the original stdin so that we can continue to receive
# input from the parent process.
stdin = sys.stdin
if sys.platform != 'win32':
try:
if sys.platform != 'darwin':
subprocess.check_call('tty', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
except (OSError, subprocess.CalledProcessError):
pass
else:
# Replace stdin with a TTY to enable pdb usage.
sys.stdin = open('/dev/tty')
while True:
input = json.loads(stdin.readline())
context = FakeLambdaContext(**input.get('context', {}))
result = handler(input['event'], context)
data = {
# just an identifier to distinguish between
# interesting data (result) and stdout/print
'__offline_payload__': result
}
sys.stdout.write(json.dumps(data))
sys.stdout.write('\n')