forked from localstack/localstack
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlambda_utils.py
More file actions
92 lines (78 loc) · 3.12 KB
/
lambda_utils.py
File metadata and controls
92 lines (78 loc) · 3.12 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
import logging
import os
from collections import defaultdict
from typing import Dict, List, Union
from localstack.utils.aws.aws_models import LambdaFunction
from localstack.utils.common import to_str
LOG = logging.getLogger(__name__)
# Lambda runtime constants
LAMBDA_RUNTIME_PYTHON36 = "python3.6"
LAMBDA_RUNTIME_PYTHON37 = "python3.7"
LAMBDA_RUNTIME_PYTHON38 = "python3.8"
LAMBDA_RUNTIME_NODEJS = "nodejs"
LAMBDA_RUNTIME_NODEJS43 = "nodejs4.3"
LAMBDA_RUNTIME_NODEJS610 = "nodejs6.10"
LAMBDA_RUNTIME_NODEJS810 = "nodejs8.10"
LAMBDA_RUNTIME_NODEJS10X = "nodejs10.x"
LAMBDA_RUNTIME_NODEJS12X = "nodejs12.x"
LAMBDA_RUNTIME_NODEJS14X = "nodejs14.x"
LAMBDA_RUNTIME_JAVA8 = "java8"
LAMBDA_RUNTIME_JAVA8_AL2 = "java8.al2"
LAMBDA_RUNTIME_JAVA11 = "java11"
LAMBDA_RUNTIME_DOTNETCORE2 = "dotnetcore2.0"
LAMBDA_RUNTIME_DOTNETCORE21 = "dotnetcore2.1"
LAMBDA_RUNTIME_DOTNETCORE31 = "dotnetcore3.1"
LAMBDA_RUNTIME_GOLANG = "go1.x"
LAMBDA_RUNTIME_RUBY = "ruby"
LAMBDA_RUNTIME_RUBY25 = "ruby2.5"
LAMBDA_RUNTIME_RUBY27 = "ruby2.7"
LAMBDA_RUNTIME_PROVIDED = "provided"
# default handler and runtime
LAMBDA_DEFAULT_HANDLER = "handler.handler"
LAMBDA_DEFAULT_RUNTIME = LAMBDA_RUNTIME_PYTHON37
LAMBDA_DEFAULT_STARTING_POSITION = "LATEST"
# List of Dotnet Lambda runtime names
DOTNET_LAMBDA_RUNTIMES = [
LAMBDA_RUNTIME_DOTNETCORE2,
LAMBDA_RUNTIME_DOTNETCORE21,
LAMBDA_RUNTIME_DOTNETCORE31,
]
def multi_value_dict_for_list(elements: List) -> Dict:
temp_mv_dict = defaultdict(list)
for key in elements:
if isinstance(key, (list, tuple)):
key, value = key
else:
value = elements[key]
key = to_str(key)
temp_mv_dict[key].append(value)
return dict((k, tuple(v)) for k, v in temp_mv_dict.items())
def get_lambda_runtime(runtime_details: Union[LambdaFunction, str]) -> str:
"""Return the runtime string from the given LambdaFunction (or runtime string)."""
if isinstance(runtime_details, LambdaFunction):
runtime_details = runtime_details.runtime
if not isinstance(runtime_details, str):
LOG.info("Unable to determine Lambda runtime from parameter: %s", runtime_details)
return runtime_details or ""
def is_provided_runtime(runtime_details: Union[LambdaFunction, str]) -> bool:
"""Whether the given LambdaFunction uses a 'provided' runtime."""
runtime = get_lambda_runtime(runtime_details) or ""
return runtime.startswith("provided")
def get_handler_file_from_name(handler_name: str, runtime: str = None):
runtime = runtime or LAMBDA_DEFAULT_RUNTIME
if runtime.startswith(LAMBDA_RUNTIME_PROVIDED):
return "bootstrap"
delimiter = "."
if runtime.startswith(LAMBDA_RUNTIME_NODEJS):
file_ext = ".js"
elif runtime.startswith(LAMBDA_RUNTIME_GOLANG):
file_ext = ""
elif runtime.startswith(tuple(DOTNET_LAMBDA_RUNTIMES)):
file_ext = ".dll"
delimiter = ":"
elif runtime.startswith(LAMBDA_RUNTIME_RUBY):
file_ext = ".rb"
else:
handler_name = handler_name.rpartition(delimiter)[0].replace(delimiter, os.path.sep)
file_ext = ".py"
return "%s%s" % (handler_name.split(delimiter)[0], file_ext)