forked from localstack/localstack
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlambda_starter.py
More file actions
50 lines (38 loc) · 1.46 KB
/
lambda_starter.py
File metadata and controls
50 lines (38 loc) · 1.46 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
import logging
from localstack import config
LOG = logging.getLogger(__name__)
def start_lambda(port=None, asynchronous=False):
from localstack.services.awslambda import lambda_api
from localstack.services.infra import start_local_api
port = port or config.service_port("lambda")
return start_local_api(
"Lambda", port, api="lambda", method=lambda_api.serve, asynchronous=asynchronous
)
def stop_lambda() -> None:
from localstack.services.awslambda.lambda_api import cleanup
"""
Stops / cleans up the Lambda Executor
"""
# TODO actually stop flask server
cleanup()
def check_lambda(expect_shutdown=False, print_error=False):
out = None
try:
from localstack.services.infra import PROXY_LISTENERS
from localstack.utils.aws import aws_stack
from localstack.utils.common import wait_for_port_open
# wait for port to be opened
# TODO get lambda port in a cleaner way
port = PROXY_LISTENERS.get("lambda")[1]
wait_for_port_open(port, sleep_time=0.5, retries=20)
endpoint_url = f"http://127.0.0.1:{port}"
out = aws_stack.connect_to_service(
service_name="lambda", endpoint_url=endpoint_url
).list_functions()
except Exception:
if print_error:
LOG.exception("Lambda health check failed")
if expect_shutdown:
assert out is None
else:
assert out and isinstance(out.get("Functions"), list)