-
Notifications
You must be signed in to change notification settings - Fork 14
/
_lambda.py
140 lines (115 loc) · 4.86 KB
/
_lambda.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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
import logging
import click
import textwrap
import copy
from pathlib import Path
from .. import files
from .component import Component
from cirrus.core.utils.yaml import NamedYamlable
logger = logging.getLogger(__name__)
class Lambda(Component):
handler = files.PythonHandler()
definition = files.LambdaDefinition()
# TODO: Readme should be required once we have one per task
readme = files.Readme(optional=True)
def load_config(self):
super().load_config()
# we only support batch on tasks, but some things are
# simpler if we know we are batch disabled for all Lambdas
self.batch_enabled = False
self.description = self.config.get('description', '')
self.environment = self.config.get('environment', NamedYamlable())
self.lambda_config = self.config.get('lambda', NamedYamlable())
self.lambda_enabled = self.lambda_config.pop('enabled', True) and self._enabled and bool(self.lambda_config)
self.lambda_config.description = self.description
project_reqs = []
if self.project and self.project.config:
project_reqs = list(
self.project.config.custom.pythonRequirements.include
)
# update task env with defaults from project config
self.environment = (
self.project.config.provider.environment | self.environment
)
# update lambda env with the merged project/task env
self.lambda_config.environment = (
self.environment | self.lambda_config.get('environment', {})
)
self.lambda_config.package = {}
self.lambda_config.package.include = []
self.lambda_config.package.include.append(f'./lambdas/{self.name}/**')
if not hasattr(self.lambda_config, 'pythonRequirements'):
self.lambda_config.pythonRequirements = {}
# note the set to deduplicate requirements
# TODO: multiple versions of the same requirement
# will not be deduplicated
self.lambda_config.pythonRequirements['include'] = sorted(list({
req for req in
# list of all requirements specified in lambda config
# and the global pythonRequiments from cirrus.yml
self.lambda_config.pythonRequirements.get('include', [])
+ project_reqs
}))
# if this is a non-container lambda that needs to be packaged
# by serverless, then we need to ensure the module points to the
# place in the build dir where we'll copy all the code
if hasattr(self.lambda_config, 'handler'):
self.lambda_config.module = f'lambdas/{self.name}'
@property
def enabled(self):
return self._enabled and (self.lambda_enabled or self.batch_enabled)
def display_attrs(self):
if self.enabled and not self.lambda_enabled and not self.batch_enabled:
yield 'DISABLED'
yield from super().display_attrs()
def detail_display(self):
super().detail_display()
click.echo(f'\nLambda enabled: {self.lambda_enabled}')
if not self.lambda_config:
return
click.echo('Lambda config:')
click.echo(textwrap.indent(self.lambda_config.to_yaml(), ' '))
def copy_for_config(self):
'''any modifications to config for serverless.yml go here'''
lc = copy.deepcopy(self.lambda_config)
lc.pop('pythonRequirements', None)
return lc
def get_outdir(self, project_build_dir: Path) -> Path:
return project_build_dir.joinpath(self.lambda_config.module) if self.lambda_enabled else None
def copy_to_outdir(self, outdir: Path) -> None:
import shutil
if not self.lambda_enabled:
return
try:
outdir.mkdir(parents=True)
except FileExistsError:
self.clean_outdir(outdir)
for _file in self.path.iterdir():
if _file.name == self.definition.name:
logger.debug('Skipping linking definition file')
continue
if _file.is_dir():
shutil.copytree(
_file,
outdir.joinpath(_file.name),
ignore=shutil.ignore_patterns('*.pyc', '__pycache__'),
)
else:
shutil.copyfile(_file, outdir.joinpath(_file.name))
outdir.joinpath('requirements.txt').write_text(
''.join(
[f'{req}\n' for req in
self.lambda_config.pythonRequirements.get('include', [])],
),
)
def clean_outdir(self, outdir: Path):
import shutil
try:
contents = outdir.iterdir()
except FileNotFoundError:
return
for _file in contents:
if not _file.is_symlink() and _file.is_dir():
shutil.rmtree(_file)
else:
_file.unlink()