-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathpublish.py
More file actions
executable file
·3224 lines (2810 loc) · 132 KB
/
publish.py
File metadata and controls
executable file
·3224 lines (2810 loc) · 132 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
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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0
"""
Create new Cfn artifacts bucket if not already existing
Build artifacts
Upload artifacts to S3 bucket for deployment with CloudFormation
"""
import concurrent.futures
import hashlib
import json
import os
import py_compile
import shutil
import subprocess
import sys
import time
import traceback
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import quote
import boto3
import yaml
from boto3.s3.transfer import TransferConfig
from botocore.exceptions import ClientError
from rich.console import Console
from rich.progress import (
Progress,
SpinnerColumn,
TextColumn,
TimeElapsedColumn,
)
LIB_DEPENDENCY = "./lib/idp_common_pkg/idp_common"
LIB_PKG_PATH = "./lib/idp_common_pkg"
class IDPPublisher:
def __init__(self, verbose=False):
self.console = Console()
self.verbose = verbose
self.bucket_basename = None
self.prefix = None
self.region = None
self.acl = None
self.bucket = None
self.prefix_and_version = None
self.version = None
self.build_errors = [] # Track build errors for verbose reporting
self.public_sample_udop_model = ""
self.public = False
self.main_template = "idp-main.yaml"
self.use_container_flag = ""
self.pattern2_use_containers = True # Default to containers for Pattern-2
self.s3_client = None
self.cf_client = None
self.sts_client = None
self._is_lib_changed = False
self.skip_validation = False
self.lint_enabled = True
self.account_id = None
self._layer_arns = {} # Store built layer ARNs for template injection
def clean_checksums(self):
"""Delete all .checksum files and Lambda layer caches for full rebuild"""
self.console.print(
"[yellow]🧹 Cleaning build cache for full rebuild...[/yellow]"
)
checksum_paths = [
".checksum", # main
"lib/.checksum", # lib
]
# Add nested stack checksum files
nested_dir = "nested"
if os.path.exists(nested_dir):
for item in os.listdir(nested_dir):
nested_path = os.path.join(nested_dir, item)
if os.path.isdir(nested_path):
checksum_paths.append(f"{nested_path}/.checksum")
# Add patterns checksum files
patterns_dir = "patterns"
if os.path.exists(patterns_dir):
for item in os.listdir(patterns_dir):
pattern_path = os.path.join(patterns_dir, item)
if os.path.isdir(pattern_path):
checksum_paths.append(f"{pattern_path}/.checksum")
deleted_count = 0
for checksum_path in checksum_paths:
if os.path.exists(checksum_path):
os.remove(checksum_path)
self.console.print(f"[green] ✓ Deleted {checksum_path}[/green]")
deleted_count += 1
# Delete cached Lambda layer zips to force layer rebuilds
layers_dir = ".aws-sam/layers"
if os.path.exists(layers_dir):
layer_zips = [f for f in os.listdir(layers_dir) if f.endswith(".zip")]
for layer_zip in layer_zips:
layer_path = os.path.join(layers_dir, layer_zip)
os.remove(layer_path)
self.console.print(f"[green] ✓ Deleted {layer_path}[/green]")
deleted_count += 1
if deleted_count == 0:
self.console.print("[dim] No cache files found to delete[/dim]")
else:
self.console.print(
f"[green]✅ Deleted {deleted_count} cache files - full rebuild will be triggered[/green]"
)
def _find_all_requirements_files(self):
"""Find all requirements.txt files in the project"""
requirements_files = []
# Main Lambda functions
src_lambda_dir = Path("src/lambda")
if src_lambda_dir.exists():
for func_dir in src_lambda_dir.iterdir():
req_file = func_dir / "requirements.txt"
if req_file.exists():
requirements_files.append(str(req_file))
# Nested Lambda functions
nested_dir = Path("nested")
if nested_dir.exists():
for nested_item in nested_dir.iterdir():
nested_src = nested_item / "src"
if nested_src.exists():
for func_dir in nested_src.iterdir():
req_file = func_dir / "requirements.txt"
if req_file.exists():
requirements_files.append(str(req_file))
# Pattern Lambda functions
patterns_dir = Path("patterns")
if patterns_dir.exists():
for pattern_dir in patterns_dir.iterdir():
pattern_src = pattern_dir / "src"
if pattern_src.exists():
for func_dir in pattern_src.iterdir():
req_file = func_dir / "requirements.txt"
if req_file.exists():
requirements_files.append(str(req_file))
return requirements_files
def _prepare_for_build_at_start(self):
"""Run at script startup - placeholder for future startup checks"""
self.log_verbose("✅ Build startup checks complete")
def log_verbose(self, message, style="dim"):
"""Log verbose messages if verbose mode is enabled"""
if self.verbose:
# Use markup=False to prevent Rich from eating brackets like [extras]
self.console.print(message, style=style, markup=False)
# ========================================================================
# LOGGING HELPERS - Consistent styling for all output
# ========================================================================
def log_phase(self, title, emoji=""):
"""Print a major phase header with separators"""
separator = "═" * 65
self.console.print(f"\n[bold cyan]{separator}[/bold cyan]")
if emoji:
self.console.print(f"[bold cyan] {emoji} {title.upper()}[/bold cyan]")
else:
self.console.print(f"[bold cyan] {title.upper()}[/bold cyan]")
self.console.print(f"[bold cyan]{separator}[/bold cyan]")
def log_task(self, message, thread=None):
"""Print task start (cyan with arrow)"""
prefix = f"[{thread}] " if thread else ""
self.console.print(f"[cyan]▶ {prefix}{message}[/cyan]")
def log_detail(self, message, thread=None):
"""Print indented detail info (dim)"""
prefix = f"[{thread}] " if thread else ""
self.console.print(f"[dim] └─ {prefix}{message}[/dim]")
def log_success(self, message, thread=None):
"""Print success message (green checkmark)"""
prefix = f"[{thread}] " if thread else ""
self.console.print(f"[green]✓ {prefix}{message}[/green]")
def log_cached(self, message, thread=None):
"""Print cached/skipped message (blue arrow)"""
prefix = f"[{thread}] " if thread else ""
self.console.print(f"[blue]→ {prefix}{message}[/blue]")
def log_warning(self, message, thread=None):
"""Print warning message (yellow)"""
prefix = f"[{thread}] " if thread else ""
self.console.print(f"[yellow]⚠ {prefix}{message}[/yellow]")
def log_error(self, message, thread=None):
"""Print error message (red X)"""
prefix = f"[{thread}] " if thread else ""
self.console.print(f"[red]✗ {prefix}{message}[/red]")
def upload_to_s3_with_timer(self, local_path, s3_key, description):
"""Upload file to S3 with a spinner, elapsed time display, and optimized transfer config.
Uses multi-threaded, multipart uploads for better performance on slow connections.
Shows progress during upload and final timing on completion.
"""
# Optimized transfer config for better upload performance
# Matches AWS CLI's optimized defaults for parallel uploads
transfer_config = TransferConfig(
multipart_threshold=5
* 1024
* 1024, # 5 MB - enable multipart for smaller files
max_concurrency=10, # Use 10 threads for parallel chunk uploads
multipart_chunksize=5 * 1024 * 1024, # 5 MB chunks
use_threads=True, # Enable multi-threading
)
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
TimeElapsedColumn(),
console=self.console,
transient=True, # Clears spinner after completion
) as progress:
progress.add_task(f"[cyan]Uploading {description}...", total=None)
start = time.time()
self.s3_client.upload_file(
local_path, self.bucket, s3_key, Config=transfer_config
)
elapsed = time.time() - start
self.log_success(f"Uploaded {description} ({elapsed:.1f}s)")
def log_error_details(self, component, error_output):
"""Log detailed error information and store for summary"""
error_info = {"component": component, "error": error_output}
self.build_errors.append(error_info)
if self.verbose:
self.console.print(f"[red]❌ {component} build failed:[/red]")
self.console.print(f"[red]{error_output}[/red]")
else:
self.console.print(
f"[red]❌ {component} build failed (use --verbose for details)[/red]"
)
def run_subprocess_with_logging(
self, cmd, component_name, cwd=None, realtime=False
):
"""Run subprocess with standardized logging"""
if realtime:
# Real-time output for long-running processes like npm install
self.console.print(f"[cyan]Running: {' '.join(cmd)}[/cyan]")
try:
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
cwd=cwd,
bufsize=1,
universal_newlines=True,
)
output_lines = []
while True:
output = process.stdout.readline()
if output == "" and process.poll() is not None:
break
if output:
line = output.strip()
output_lines.append(line)
# Show progress for npm commands
if "npm" in " ".join(cmd):
if any(
keyword in line.lower()
for keyword in [
"downloading",
"installing",
"added",
"updated",
"audited",
]
):
self.console.print(f"[dim] {line}[/dim]")
elif "warn" in line.lower():
self.console.print(f"[yellow] {line}[/yellow]")
elif "error" in line.lower():
self.console.print(f"[red] {line}[/red]")
return_code = process.poll()
if return_code != 0:
error_msg = f"""Command failed: {" ".join(cmd)}
Working directory: {cwd or os.getcwd()}
Return code: {return_code}
OUTPUT:
{chr(10).join(output_lines)}"""
print(error_msg)
self.log_error_details(component_name, error_msg)
return False, error_msg
return True, None # Success, no result object needed for real-time
except Exception as e:
error_msg = (
f"Failed to execute command: {' '.join(cmd)}\nError: {str(e)}"
)
self.log_error_details(component_name, error_msg)
return False, error_msg
else:
# Original behavior - capture all output
result = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
if result.returncode != 0:
error_msg = f"""Command failed: {" ".join(cmd)}
Working directory: {cwd or os.getcwd()}
Return code: {result.returncode}
STDOUT:
{result.stdout}
STDERR:
{result.stderr}"""
print(error_msg)
self.log_error_details(component_name, error_msg)
return False, error_msg
return True, result
def print_error_summary(self):
"""Print summary of all build errors"""
if not self.build_errors:
return
self.console.print("\n[red]❌ Build Error Summary:[/red]")
for i, error_info in enumerate(self.build_errors, 1):
self.console.print(f"\n[red]{i}. {error_info['component']}:[/red]")
if self.verbose:
self.console.print(f"[red]{error_info['error']}[/red]")
else:
# Show first few lines of error for non-verbose mode
error_lines = error_info["error"].strip().split("\n")
preview_lines = error_lines[:3] # Show first 3 lines
for line in preview_lines:
self.console.print(f"[red] {line}[/red]")
if len(error_lines) > 3:
self.console.print(
f"[dim] ... ({len(error_lines) - 3} more lines, use --verbose for full output)[/dim]"
)
def print_usage(self):
"""Print usage information with Rich formatting"""
self.console.print("\n[bold cyan]Usage:[/bold cyan]")
self.console.print(
" python3 publish.py <cfn_bucket_basename> <cfn_prefix> <region> [public] [--max-workers N] [--verbose] [--no-validate] [--lint on|off]"
)
self.console.print("\n[bold cyan]Parameters:[/bold cyan]")
self.console.print(
" [yellow]<cfn_bucket_basename>[/yellow]: Base name for the CloudFormation artifacts bucket"
)
self.console.print(" [yellow]<cfn_prefix>[/yellow]: S3 prefix for artifacts")
self.console.print(" [yellow]<region>[/yellow]: AWS region for deployment")
self.console.print(
" [yellow][public][/yellow]: Optional. If 'public', artifacts will be made publicly readable"
)
self.console.print(
" [yellow][--max-workers N][/yellow]: Optional. Maximum number of concurrent workers (default: auto-detect)"
)
self.console.print(
" Use 1 for sequential processing, higher numbers for more concurrency"
)
self.console.print(
" [yellow][--verbose, -v][/yellow]: Optional. Enable verbose output for debugging"
)
self.console.print(
" [yellow][--no-validate][/yellow]: Optional. Skip CloudFormation template validation"
)
self.console.print(
" [yellow][--clean-build][/yellow]: Optional. Delete all .checksum files to force full rebuild"
)
self.console.print(
" [yellow][--lint on|off][/yellow]: Optional. Enable/disable UI linting and build validation (default: on)"
)
def check_parameters(self, args):
"""Check and validate input parameters"""
if len(args) < 3:
self.console.print("[red]Error: Missing required parameters[/red]")
self.print_usage()
sys.exit(1)
# Parse arguments
self.bucket_basename = args[0]
self.prefix = args[1].rstrip("/") # Remove trailing slash
self.region = args[2]
# Default values
self.public = False
self.acl = "bucket-owner-full-control"
self.max_workers = None # Auto-detect
# Parse optional arguments
remaining_args = args[3:]
i = 0
while i < len(remaining_args):
arg = remaining_args[i]
if arg.lower() == "public":
self.public = True
self.acl = "public-read"
self.console.print(
"[green]Published S3 artifacts will be accessible by public.[/green]"
)
elif arg == "--max-workers":
if i + 1 >= len(remaining_args):
self.console.print(
"[red]Error: --max-workers requires a number[/red]"
)
self.print_usage()
sys.exit(1)
try:
self.max_workers = int(remaining_args[i + 1])
if self.max_workers < 1:
self.console.print(
"[red]Error: --max-workers must be at least 1[/red]"
)
sys.exit(1)
self.console.print(
f"[green]Using {self.max_workers} concurrent workers[/green]"
)
i += 1 # Skip the next argument (the number)
except ValueError:
self.console.print(
"[red]Error: --max-workers must be followed by a valid number[/red]"
)
self.print_usage()
sys.exit(1)
elif arg in ["--verbose", "-v"]:
self.verbose = True
self.console.print("[green]Verbose mode enabled[/green]")
elif arg == "--no-validate":
self.skip_validation = True
self.console.print(
"[yellow]CloudFormation template validation will be skipped[/yellow]"
)
elif arg == "--lint":
if i + 1 >= len(remaining_args):
self.console.print(
"[red]Error: --lint requires 'on' or 'off'[/red]"
)
self.print_usage()
sys.exit(1)
lint_value = remaining_args[i + 1].lower()
if lint_value not in ["on", "off"]:
self.console.print("[red]Error: --lint must be 'on' or 'off'[/red]")
self.print_usage()
sys.exit(1)
self.lint_enabled = lint_value == "on"
i += 1 # increment arg counter to avoid parsing "on/off" as an arg of its own
elif arg == "--clean-build":
self.clean_checksums()
else:
self.console.print(
f"[yellow]Warning: Unknown argument '{arg}' ignored[/yellow]"
)
i += 1
if not self.public:
self.console.print(
"[yellow]Published S3 artifacts will NOT be accessible by public.[/yellow]"
)
def setup_environment(self):
"""Set up environment variables and derived values"""
os.environ["AWS_DEFAULT_REGION"] = self.region
# Initialize AWS clients
self.s3_client = boto3.client("s3", region_name=self.region)
self.cf_client = boto3.client("cloudformation", region_name=self.region)
# Read version
try:
with open("./VERSION", "r") as f:
self.version = f.read().strip()
except FileNotFoundError:
self.console.print("[red]Error: VERSION file not found[/red]")
sys.exit(1)
self.prefix_and_version = f"{self.prefix}/{self.version}"
self.bucket = f"{self.bucket_basename}-{self.region}"
# Set UDOP model path based on region
self.public_sample_udop_model = f"s3://aws-ml-blog-{self.region}/artifacts/genai-idp/udop-finetuning/rvl-cdip/model.tar.gz"
def check_prerequisites(self):
"""Check for required commands and versions"""
# Check required commands
required_commands = ["aws", "sam"]
for cmd in required_commands:
if not shutil.which(cmd):
self.console.print(
f"[red]Error: {cmd} is required but not installed[/red]"
)
sys.exit(1)
# Check SAM version
try:
result = subprocess.run(
["sam", "--version"], capture_output=True, text=True, check=True
)
sam_version = result.stdout.split()[3] # Extract version from output
min_sam_version = "1.129.0"
if self.version_compare(sam_version, min_sam_version) < 0:
self.console.print(
f"[red]Error: sam version >= {min_sam_version} is required. (Installed version is {sam_version})[/red]"
)
self.console.print(
"[yellow]Install: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/manage-sam-cli-versions.html[/yellow]"
)
sys.exit(1)
except subprocess.CalledProcessError:
self.console.print("[red]Error: Could not determine SAM version[/red]")
sys.exit(1)
# Check Python version
python_version = f"{sys.version_info.major}.{sys.version_info.minor}"
min_python_version = "3.12"
if self.version_compare(python_version, min_python_version) < 0:
self.console.print(
f"[red]Error: Python version >= {min_python_version} is required. (Installed version is {python_version})[/red]"
)
sys.exit(1)
def version_compare(self, version1, version2):
"""Compare two version strings. Returns -1 if v1 < v2, 0 if equal, 1 if v1 > v2"""
def normalize(v):
return [int(x) for x in v.split(".")]
v1_parts = normalize(version1)
v2_parts = normalize(version2)
# Pad shorter version with zeros
max_len = max(len(v1_parts), len(v2_parts))
v1_parts.extend([0] * (max_len - len(v1_parts)))
v2_parts.extend([0] * (max_len - len(v2_parts)))
for i in range(max_len):
if v1_parts[i] < v2_parts[i]:
return -1
elif v1_parts[i] > v2_parts[i]:
return 1
return 0
def setup_artifacts_bucket(self):
"""Create bucket if necessary"""
try:
self.s3_client.head_bucket(Bucket=self.bucket)
self.console.print(f"[green]Using existing bucket: {self.bucket}[/green]")
except ClientError as e:
error_code = e.response["Error"]["Code"]
if error_code == "404":
self.console.print(
f"[yellow]Creating s3 bucket: {self.bucket}[/yellow]"
)
try:
if self.region == "us-east-1":
self.s3_client.create_bucket(Bucket=self.bucket)
else:
self.s3_client.create_bucket(
Bucket=self.bucket,
CreateBucketConfiguration={
"LocationConstraint": self.region
},
)
# Enable versioning
self.s3_client.put_bucket_versioning(
Bucket=self.bucket,
VersioningConfiguration={"Status": "Enabled"},
)
except ClientError as create_error:
self.console.print(
f"[red]Failed to create bucket: {create_error}[/red]"
)
sys.exit(1)
else:
self.console.print("[red]Error accessing bucket:[/red]")
self.console.print(str(e), style="red", markup=False)
sys.exit(1)
def get_file_checksum(self, file_path):
"""Get SHA256 checksum of a file"""
if not os.path.exists(file_path):
return ""
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
def get_directory_checksum(self, directory):
"""Get combined checksum of all files in a directory, excluding development artifacts"""
if not os.path.exists(directory):
return ""
# Define patterns to exclude from checksum calculation
exclude_dirs = {
"__pycache__",
".pytest_cache",
".ruff_cache",
"build",
"dist",
".aws-sam",
"node_modules",
".git",
".vscode",
".idea",
"test-reports", # Exclude test report directories
}
exclude_file_patterns = {
".checksum",
".build_checksum",
"lib/.checksum",
".pyc",
".pyo",
".pyd",
".so",
".egg-info",
".coverage",
".DS_Store",
"Thumbs.db",
"coverage.xml", # Coverage reports
"test-results.xml", # Test result reports
".gitkeep", # Git placeholder files
}
exclude_file_suffixes = (
".pyc",
".pyo",
".pyd",
".so",
".coverage",
".log", # Log files
)
exclude_dir_suffixes = (".egg-info",)
def should_exclude_dir(dir_name):
"""Check if directory should be excluded from checksum"""
if dir_name in exclude_dirs:
return True
if any(dir_name.endswith(suffix) for suffix in exclude_dir_suffixes):
return True
# Exclude test directories for library checksum only
if "lib" in directory and (
dir_name == "tests" or dir_name.startswith("test_")
):
return True
return False
def should_exclude_file(file_name):
"""Check if file should be excluded from checksum"""
if file_name in exclude_file_patterns:
return True
if any(file_name.endswith(suffix) for suffix in exclude_file_suffixes):
return True
# Exclude test files for library checksum only
if "lib" in directory and (
file_name.startswith("test_")
or file_name.endswith("_test.py")
or file_name == "nodeids" # pytest cache files
or file_name == "lastfailed" # pytest cache files
or file_name
in ["coverage.xml", "test-results.xml"] # specific test report files
):
return True
return False
checksums = []
for root, dirs, files in os.walk(directory):
# Filter out excluded directories in-place to prevent os.walk from descending into them
dirs[:] = [d for d in dirs if not should_exclude_dir(d)]
# Sort to ensure consistent ordering
dirs.sort()
files.sort()
for file in files:
if not should_exclude_file(file):
file_path = os.path.join(root, file)
if os.path.isfile(file_path):
checksums.append(self.get_file_checksum(file_path))
# Combine all checksums
combined = "".join(checksums)
return hashlib.sha256(combined.encode()).hexdigest()
def build_and_package_template(self, directory, force_rebuild=False):
"""Build and package a template directory with smart rebuild detection"""
# Track build time
build_start = time.time()
try:
# Pattern-2 uses containers - images built separately by build_and_push_pattern2_containers()
# SAM build with SkipBuild: True just prepares template
cmd = ["sam", "build", "--template-file", "template.yaml"]
# Add container flag if needed
if self.use_container_flag and self.use_container_flag.strip():
cmd.append(self.use_container_flag)
if self.verbose:
cmd.append("--debug")
sam_build_start = time.time()
# Validate Python syntax before building
if not self._validate_python_syntax(directory):
raise Exception("Python syntax validation failed")
self.log_verbose(
f"Running SAM build command in {directory}: {' '.join(cmd)}"
)
# Run SAM build from the pattern directory
success, result = self.run_subprocess_with_logging(
cmd, f"SAM build for {directory}", directory
)
sam_build_time = time.time() - sam_build_start
if not success:
raise Exception("SAM build failed")
# Package the template (using absolute paths)
build_template_path = os.path.join(
directory, ".aws-sam", "build", "template.yaml"
)
# Use different name for pattern-2 container deployment
if directory == "patterns/pattern-2" and self.pattern2_use_containers:
packaged_template_path = os.path.join(
directory, ".aws-sam", "packaged-container.yaml"
)
else:
# Use standard packaged.yaml name
packaged_template_path = os.path.join(
directory, ".aws-sam", "packaged.yaml"
)
cmd = [
"sam",
"package",
"--template-file",
build_template_path,
"--output-template-file",
packaged_template_path,
"--s3-bucket",
self.bucket,
"--s3-prefix",
self.prefix_and_version,
]
if self.verbose:
cmd.append("--debug")
# Pattern-1, Pattern-2, and Pattern-3 need --image-repository even with SkipBuild: True
# SAM package uses this to generate correct ImageUri references in the template
if directory in ["patterns/pattern-1", "patterns/pattern-3"] or (
directory == "patterns/pattern-2" and self.pattern2_use_containers
):
placeholder_ecr = (
f"{self.account_id}.dkr.ecr.{self.region}.amazonaws.com/placeholder"
)
cmd.extend(["--image-repository", placeholder_ecr])
sam_package_start = time.time()
self.log_verbose(f"Running SAM package command: {' '.join(cmd)}")
# Run SAM package from project root (no cwd change needed)
success, result = self.run_subprocess_with_logging(
cmd, f"SAM package for {directory}"
)
sam_package_time = time.time() - sam_package_start
if not success:
raise Exception("SAM package failed")
# For Pattern-2 with containers, ensure packaged.yaml exists with standard name
if directory == "patterns/pattern-2" and self.pattern2_use_containers:
standard_packaged_path = os.path.join(
directory, ".aws-sam", "packaged.yaml"
)
# If using a different packaged name, copy to standard name for main template compatibility
if packaged_template_path != standard_packaged_path:
import shutil
shutil.copy2(packaged_template_path, standard_packaged_path)
self.log_verbose(
"Created packaged.yaml copy for Pattern-2 compatibility"
)
# Log S3 upload location for Lambda artifacts
self.console.print(
f"[dim] 📤 Lambda artifacts uploaded to s3://{self.bucket}/{self.prefix_and_version}/[/dim]"
)
# Log timing information
total_time = time.time() - build_start
pattern_name = os.path.basename(directory)
self.console.print(
f"[dim] {pattern_name}: build={sam_build_time:.1f}s, package={sam_package_time:.1f}s, total={total_time:.1f}s[/dim]"
)
except Exception as e:
# Delete checksum on any failure to force rebuild next time
self._delete_checksum_file(directory)
self.log_verbose(f"Exception in build_and_package_template: {str(e)}")
self.log_verbose(f"Traceback: {traceback.format_exc()}")
self.console.print(f"[red]❌ Build failed for {directory}:[/red]")
self.console.print(str(e), style="red", markup=False)
sys.exit(1)
return True
def build_components_with_smart_detection(
self, components_needing_rebuild, component_type, max_workers=None
):
"""Build patterns or options with smart detection using Lambda Layers."""
# Filter components by type
components_to_build = []
for item in components_needing_rebuild:
if component_type in item["component"]:
components_to_build.append(item["component"])
if not components_to_build:
self.console.print(f"[green]✅ All {component_type} are up to date[/green]")
return True
self.console.print(
f"[cyan]Building {len(components_to_build)} {component_type} with {max_workers} workers...[/cyan]"
)
return self._build_components_concurrently(
components_to_build, component_type, max_workers
)
def _build_components_concurrently(self, components, component_type, max_workers):
"""Generic method to build components concurrently with simple logging.
Note: Progress bars removed to avoid Rich LiveDisplay conflicts when building
categories concurrently. Simple status logging used instead.
"""
# Use ThreadPoolExecutor for I/O bound operations (sam build/package)
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Submit all component build tasks
future_to_component = {}
for component in components:
self.log_task("Building...", thread=component)
future = executor.submit(
self.build_and_package_template, component, force_rebuild=True
)
future_to_component[future] = component
# Wait for all tasks to complete and check results
all_successful = True
completed = 0
for future in concurrent.futures.as_completed(future_to_component):
component = future_to_component[future]
completed += 1
try:
success = future.result()
if not success:
self.log_error("Build failed!", thread=component)
all_successful = False
else:
self.log_success(
f"Complete ({completed}/{len(components)})",
thread=component,
)
except Exception as e:
# Log detailed error information
error_output = (
f"Exception: {str(e)}\n\nTraceback:\n{traceback.format_exc()}"
)
self.log_error_details(
f"{component_type.title()} {component} build exception",
error_output,
)
self.log_error(f"Error: {str(e)[:50]}...", thread=component)
all_successful = False
return all_successful
def generate_config_file_list(self):
"""Generate list of configuration files for explicit copying"""
config_dir = "config_library"
file_list = []
for root, dirs, files in os.walk(config_dir):
for file in files:
file_path = os.path.join(root, file)
relative_path = os.path.relpath(file_path, config_dir)
file_list.append(relative_path)
return sorted(file_list)
def _extract_function_name(self, dir_name, template_path):
"""Extract CloudFormation function name from template by matching CodeUri."""
try:
# Create a custom loader that ignores CloudFormation intrinsic functions
class CFLoader(yaml.SafeLoader):
pass
def construct_unknown(loader, node):
if isinstance(node, yaml.ScalarNode):
return loader.construct_scalar(node)
elif isinstance(node, yaml.SequenceNode):
return loader.construct_sequence(node)
elif isinstance(node, yaml.MappingNode):
return loader.construct_mapping(node)
return None
# Add constructors for CloudFormation intrinsic functions
cf_functions = [
"!Ref",
"!GetAtt",
"!Join",
"!Sub",
"!Select",
"!Split",
"!Base64",
"!GetAZs",
"!ImportValue",
"!FindInMap",
"!Equals",
"!And",
"!Or",
"!Not",
"!If",
"!Condition",
]
for func in cf_functions:
CFLoader.add_constructor(func, construct_unknown)
with open(template_path, "r", encoding="utf-8") as f:
template = yaml.load(f, Loader=CFLoader)
if not template or not isinstance(template, dict):
raise Exception(f"Failed to parse YAML template: {template_path}")
resources = template.get("Resources", {})
for resource_name, resource_config in resources.items():
if (
resource_config
and isinstance(resource_config, dict)
and resource_config.get("Type") == "AWS::Serverless::Function"
):
properties = resource_config.get("Properties", {})
if properties and isinstance(properties, dict):
code_uri = properties.get("CodeUri", "")
if isinstance(code_uri, str):
code_uri = code_uri.rstrip("/")
code_dir = (
code_uri.split("/")[-1] if "/" in code_uri else code_uri
)
if code_dir == dir_name:
return resource_name
raise Exception(
f"No CloudFormation function found for directory {dir_name} in template {template_path}"
)
except Exception as e:
self.console.print(
f"[yellow]⚠ Warning: Could not extract function name for {dir_name} from {template_path}:[/yellow]"
)
self.console.print(f"[dim]{str(e)}[/dim]")
# Don't exit - just skip this function
return None
def upload_config_library(self):
"""Upload configuration library to S3 using aws s3 sync.
Uses AWS CLI's built-in concurrency and delta sync for optimal performance.
AWS CLI automatically skips unchanged files and uses parallel uploads.
"""
self.log_phase("Uploading Config Library", "📂")