Coverage for scripts/build_layer.py: 95%

1"""Build a Lambda layer zip containing the async durable execution SDK.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import importlib.util 

7import logging 

8import subprocess 

9import sys 

10import tempfile 

11import zipfile 

12from dataclasses import dataclass 

13from importlib.metadata import PackageNotFoundError 

14from importlib.metadata import version 

15from pathlib import Path 

16from typing import Sequence 

17 

18 

19logger = logging.getLogger(__name__) 

20DEFAULT_OUTPUT = Path("dist") / "async-durable-execution-layer.zip" 

21LAYER_ROOT = "python" 

22 

23 

24@dataclass(frozen=True) 

25class LayerBuildResult: 

26 """Details about a generated Lambda layer archive.""" 

27 

28 output_path: Path 

29 sdk_source: str 

30 file_count: int 

31 size_bytes: int 

32 

33 

34def default_sdk_spec() -> str: 

35 """Return the SDK package spec matching this layer package version.""" 

36 

37 version = _read_packaged_version() 

38 return f"async-durable-execution=={version}" 

39 

40 

41def build_layer( 

42 *, 

43 output_path: Path = DEFAULT_OUTPUT, 

44 sdk_source: str | None = None, 

45 python_executable: str = sys.executable, 

46 pip_args: Sequence[str] = (), 

47) -> LayerBuildResult: 

48 """Install the SDK into a Lambda layer layout and zip it. 

49 

50 The generated archive has the standard Python layer layout: 

51 ``python/<installed packages>``. 

52 """ 

53 

54 resolved_source = sdk_source or default_sdk_spec() 

55 output_path = output_path.resolve() 

56 

57 with tempfile.TemporaryDirectory(prefix="async-durable-layer-") as temp_dir: 

58 staging_dir = Path(temp_dir) 

59 python_dir = staging_dir / LAYER_ROOT 

60 python_dir.mkdir(parents=True) 

61 

62 _pip_install( 

63 python_executable=python_executable, 

64 target_dir=python_dir, 

65 sdk_source=resolved_source, 

66 pip_args=pip_args, 

67 ) 

68 

69 file_count = create_layer_archive(staging_dir, output_path) 

70 

71 return LayerBuildResult( 

72 output_path=output_path, 

73 sdk_source=resolved_source, 

74 file_count=file_count, 

75 size_bytes=output_path.stat().st_size, 

76 ) 

77 

78 

79def create_layer_archive(staging_dir: Path, output_path: Path) -> int: 

80 """Create a deterministic zip archive from a staged Lambda layer directory.""" 

81 

82 python_dir = staging_dir / LAYER_ROOT 

83 if not python_dir.is_dir(): 

84 msg = f"Layer staging directory must contain a {LAYER_ROOT!r} directory" 

85 raise ValueError(msg) 

86 

87 output_path.parent.mkdir(parents=True, exist_ok=True) 

88 if output_path.exists(): 

89 output_path.unlink() 

90 

91 file_count = 0 

92 with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as archive: 

93 for path in sorted(staging_dir.rglob("*")): 

94 if not path.is_file(): 

95 continue 

96 if _should_skip_archive_path(path): 

97 continue 

98 archive_path = path.relative_to(staging_dir).as_posix() 

99 info = zipfile.ZipInfo(archive_path) 

100 info.date_time = (1980, 1, 1, 0, 0, 0) 

101 info.external_attr = (0o644 & 0xFFFF) << 16 

102 with path.open("rb") as file: 

103 archive.writestr(info, file.read(), compress_type=zipfile.ZIP_DEFLATED) 

104 file_count += 1 

105 

106 return file_count 

107 

108 

109def _pip_install( 

110 *, 

111 python_executable: str, 

112 target_dir: Path, 

113 sdk_source: str, 

114 pip_args: Sequence[str], 

115) -> None: 

116 command = [ 

117 python_executable, 

118 "-m", 

119 "pip", 

120 "install", 

121 "--upgrade", 

122 "--no-compile", 

123 "--target", 

124 str(target_dir), 

125 *pip_args, 

126 sdk_source, 

127 ] 

128 logger.info("Installing %s into %s", sdk_source, target_dir) 

129 subprocess.run(command, check=True) 

130 

131 

132def _should_skip_archive_path(path: Path) -> bool: 

133 return "__pycache__" in path.parts or path.suffix in {".pyc", ".pyo"} 

134 

135 

136def _read_packaged_version() -> str: 

137 source_path = Path(__file__).resolve() 

138 for parent in source_path.parents: 

139 version_path = parent / "VERSION.py" 

140 if version_path.exists(): 

141 return _load_version_from_path(version_path) 

142 

143 try: 

144 return version("async-durable-execution") 

145 except PackageNotFoundError as error: 

146 msg = "Unable to resolve async-durable-execution version" 

147 raise RuntimeError(msg) from error 

148 

149 

150def _load_version_from_path(path: Path) -> str: 

151 spec = importlib.util.spec_from_file_location("_async_durable_layer_version", path) 

152 if spec is None or spec.loader is None: 

153 msg = f"Unable to load version from {path}" 

154 raise RuntimeError(msg) 

155 

156 module = importlib.util.module_from_spec(spec) 

157 spec.loader.exec_module(module) 

158 return module.__version__ 

159 

160 

161def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: 

162 parser = argparse.ArgumentParser( 

163 description="Build an AWS Lambda layer zip for async-durable-execution" 

164 ) 

165 parser.add_argument( 

166 "--output", 

167 type=Path, 

168 default=DEFAULT_OUTPUT, 

169 help=f"Layer zip output path (default: {DEFAULT_OUTPUT})", 

170 ) 

171 parser.add_argument( 

172 "--sdk-source", 

173 default=None, 

174 help=( 

175 "Package spec or local path to install into the layer " 

176 f"(default: {default_sdk_spec()})" 

177 ), 

178 ) 

179 parser.add_argument( 

180 "--python", 

181 default=sys.executable, 

182 help="Python executable used to run pip", 

183 ) 

184 parser.add_argument( 

185 "--pip-arg", 

186 action="append", 

187 default=[], 

188 help="Additional argument to pass to pip install; repeat for multiple args", 

189 ) 

190 parser.add_argument( 

191 "--verbose", 

192 action="store_true", 

193 help="Enable verbose logging", 

194 ) 

195 return parser.parse_args(argv) 

196 

197 

198def main(argv: Sequence[str] | None = None) -> int: 

199 args = parse_args(argv) 

200 logging.basicConfig( 

201 level=logging.INFO if args.verbose else logging.WARNING, 

202 format="%(levelname)s: %(message)s", 

203 ) 

204 

205 try: 

206 result = build_layer( 

207 output_path=args.output, 

208 sdk_source=args.sdk_source, 

209 python_executable=args.python, 

210 pip_args=args.pip_arg, 

211 ) 

212 except subprocess.CalledProcessError as error: 

213 print(f"Failed to install layer dependencies: {error}", file=sys.stderr) 

214 return 1 

215 except (OSError, RuntimeError, ValueError, zipfile.BadZipFile) as error: 

216 print(f"Failed to build layer: {error}", file=sys.stderr) 

217 return 1 

218 

219 print( 

220 "Built Lambda layer " 

221 f"{result.output_path} with {result.file_count} files " 

222 f"({result.size_bytes} bytes) from {result.sdk_source}" 

223 ) 

224 return 0 

225 

226 

227if __name__ == "__main__": 

228 raise SystemExit(main())