forked from nficano/python-lambda
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathaws_lambda.py
More file actions
executable file
·98 lines (73 loc) · 2.83 KB
/
aws_lambda.py
File metadata and controls
executable file
·98 lines (73 loc) · 2.83 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
# -*- coding: utf-8 -*-
import json
import logging
import os
import time
from imp import load_source
import yaml
from .helpers import read, custom_copytree
log = logging.getLogger(__name__)
def invoke(src, alt_event=None, verbose=False):
"""Simulates a call to your function.
:param str src:
The path to your Lambda ready project (folder must contain a valid
config.yaml and handler module (e.g.: service.py).
:param str alt_event:
An optional argument to override which event file to use.
:param bool verbose:
Whether to print out verbose details.
"""
# Load and parse the config file.
path_to_config_file = os.path.join(src, 'config.yaml')
cfg = read(path_to_config_file, loader=yaml.load)
# Load and parse event file.
if alt_event:
path_to_event_file = os.path.join(src, alt_event)
else:
path_to_event_file = os.path.join(src, 'event.json')
event = read(path_to_event_file, loader=json.loads)
handler = cfg.get('handler')
# Inspect the handler string (<module>.<function name>) and translate it
# into a function we can execute.
fn = get_callable_handler_function(src, handler)
# TODO: look into mocking the ``context`` variable, currently being passed
# as None.
start = time.time()
results = fn(event, None)
end = time.time()
print("{0}".format(results))
if verbose:
print("\nexecution time: {:.8f}s\nfunction execution "
"timeout: {:2}s".format(end - start, cfg.get('timeout', 15)))
def init(src, minimal=False):
"""Copies template files to a given directory.
:param str src:
The path to output the template lambda project files.
:param bool minimal:
Minimal possible template files (excludes event.json).
"""
destination = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "project_templates")
custom_copytree(destination, src, minimal)
def get_callable_handler_function(src, handler):
"""Tranlate a string of the form "module.function" into a callable
function.
:param str src:
The path to your Lambda project containing a valid handler file.
:param str handler:
A dot delimited string representing the `<module>.<function name>`.
"""
# "cd" into `src` directory.
os.chdir(src)
module_name, function_name = handler.split('.')
filename = get_handler_filename(handler)
path_to_module_file = os.path.join(src, filename)
module = load_source(module_name, path_to_module_file)
return getattr(module, function_name)
def get_handler_filename(handler):
"""Shortcut to get the filename from the handler string.
:param str handler:
A dot delimited string representing the `<module>.<function name>`.
"""
module_name, _ = handler.split('.')
return '{0}.py'.format(module_name)