-
Notifications
You must be signed in to change notification settings - Fork 6
/
lambda_function.py
56 lines (44 loc) · 1.66 KB
/
lambda_function.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
import logging
from typing import Any, Dict
from notifier.config.local import read_local_auth, read_local_config
from notifier.main import main
# AWS Lambda runtime attaches a default handler with formatting I don't like
# - remove it
rootLogger = logging.getLogger()
if rootLogger.handlers:
for handler in rootLogger.handlers:
rootLogger.removeHandler(handler)
# Add my own formatter
logging.basicConfig(
format="%(asctime)s - %(name)s:%(lineno)s - %(levelname)s - %(message)s",
level=logging.DEBUG,
)
logger = logging.getLogger(__name__)
def lambda_handler(event: Dict[str, str], context: Any) -> int:
"""Handler for an AWS Lambda.
:param event: Event parameter passed to the lambda. Expected to contain the
paths to the config file and auth file. May also contain the start time.
"""
logger.info("Starting Lambda")
print("Handler received event:", event)
del context
if not isinstance(event, dict):
raise ValueError("Event should be a dict")
if "config_path" not in event:
raise ValueError("Missing key config_path in event")
local_config_path = event["config_path"]
if "auth_path" not in event:
raise ValueError("Missing key auth_path in event")
local_auth_path = event["auth_path"]
force_current_time = None
if "force_current_time" in event:
force_current_time = event["force_current_time"]
logger.debug("Lambda: starting main procedure")
main(
config=read_local_config(local_config_path),
auth=read_local_auth(local_auth_path),
execute_now=[],
force_current_time=force_current_time,
)
logger.info("Lambda finished")
return 0