Coverage for lambda / image-lookup / handler.py: 100.00%
100 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Lookup-or-create custom resource handler for ECR repositories.
4Handles CloudFormation custom resource events for ``gco/<name>``
5repositories. The handler implements the adopt-or-create pattern so
6that a previously retained repository (left over from a prior deploy
7with ``RemovalPolicy=RETAIN``) is rebound to the new stack rather than
8failing with ``RepositoryAlreadyExistsException``.
10Event shape (CloudFormation custom resource):
12 event["RequestType"] — "Create" | "Update" | "Delete"
13 event["ResourceProperties"]:
14 RepositoryName — full repo name like ``gco/my-app``
15 RemovalPolicy — ``"retain"`` | ``"destroy"``
16 EmptyOnDelete — ``True`` | ``False``
17 LifecyclePolicy — optional JSON string of the lifecycle policy
19Behaviour:
21 Create / Update:
22 DescribeRepositories → if found, adopt; else CreateRepository.
23 Then PutLifecyclePolicy when ``LifecyclePolicy`` is provided.
25 Delete:
26 Read tags via ListTagsForResource. If ``gco:retain=true`` is
27 set on the repo, log + return success without deleting (the
28 retain tag wins regardless of stack-level ``RemovalPolicy``).
29 Else: when ``RemovalPolicy=="destroy"`` AND
30 ``EmptyOnDelete==True``, BatchDeleteImage every image then
31 DeleteRepository. When ``RemovalPolicy=="destroy"`` AND
32 ``EmptyOnDelete==False``, DeleteRepository (which ECR rejects
33 for non-empty repos — surfaces as a CloudFormation rollback).
34 When ``RemovalPolicy=="retain"``, return success without any
35 delete call.
37The handler returns the standard CloudFormation custom resource shape
38``{"PhysicalResourceId": <repo_arn>, "Data": {...}}``. The CDK
39``Provider`` framework wraps this into the protocol-required response
40envelope when invoked through ``CustomResource``.
41"""
43from __future__ import annotations
45import json
46import logging
47from typing import Any
49import boto3
51# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
52# Generated at (UTC): 2026-09-10T23:26:44Z
53# Generated from Git commit: 4c42b84d53d6cc01cd2b3c7e4011a43f850678b6
54# Flowchart(s) generated from this file:
55# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/image-lookup/handler.lambda_handler.html``
56# (PNG: ``diagrams/code_diagrams/lambda/image-lookup/handler.lambda_handler.png``)
57# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
58# <pyflowchart-code-diagram> END
61logger = logging.getLogger()
62logger.setLevel(logging.INFO)
65def _ecr_client() -> Any:
66 """Return a region-default ECR boto3 client."""
67 return boto3.client("ecr")
70def _describe_repository(ecr: Any, repository_name: str) -> dict[str, Any] | None:
71 """Return the repository description if it exists, else ``None``.
73 ECR raises ``RepositoryNotFoundException`` when the named repository
74 does not exist; we translate that into ``None`` so the caller can
75 distinguish missing from an actual API error.
76 """
77 try:
78 resp = ecr.describe_repositories(repositoryNames=[repository_name])
79 except ecr.exceptions.RepositoryNotFoundException:
80 return None
81 except Exception as exc: # noqa: BLE001
82 # Some boto3 stubs surface RepositoryNotFoundException via the
83 # generic ClientError shape rather than the typed exception. Sniff
84 # the error code and translate consistently.
85 code = getattr(exc, "response", {}).get("Error", {}).get("Code", "")
86 if code == "RepositoryNotFoundException":
87 return None
88 raise
90 repos = resp.get("repositories", [])
91 return repos[0] if repos else None
94def _create_repository(ecr: Any, repository_name: str) -> dict[str, Any]:
95 """Create the named repository with project-standard configuration."""
96 resp = ecr.create_repository(
97 repositoryName=repository_name,
98 imageTagMutability="MUTABLE",
99 imageScanningConfiguration={"scanOnPush": True},
100 )
101 repo: dict[str, Any] = resp.get("repository", {})
102 return repo
105def _apply_lifecycle_policy(ecr: Any, repository_name: str, lifecycle_policy: str | None) -> None:
106 """Apply ``lifecycle_policy`` (a JSON string) to the repository when set.
108 Silently no-ops when the value is empty or whitespace. Validates the
109 JSON shape before calling ``put_lifecycle_policy`` so a malformed
110 policy surfaces as a custom-resource error rather than a confusing
111 ECR-side validation failure.
112 """
113 if not lifecycle_policy or not lifecycle_policy.strip():
114 return
115 # Validate the JSON parses; ``put_lifecycle_policy`` accepts the raw
116 # string, but this gives a clearer error message on invalid input.
117 json.loads(lifecycle_policy)
118 ecr.put_lifecycle_policy(
119 repositoryName=repository_name,
120 lifecyclePolicyText=lifecycle_policy,
121 )
124def _has_retain_tag(ecr: Any, repository_arn: str) -> bool:
125 """Return True when ``gco:retain=true`` is present on the repository."""
126 try:
127 resp = ecr.list_tags_for_resource(resourceArn=repository_arn)
128 except Exception as exc: # noqa: BLE001
129 logger.error("list_tags_for_resource failed for %s: %s", repository_arn, exc)
130 raise RuntimeError(
131 f"Unable to verify retention tags for {repository_arn}; refusing deletion"
132 ) from exc
133 for tag in resp.get("tags", []) or []:
134 if tag.get("Key") == "gco:retain" and str(tag.get("Value", "")).lower() == "true":
135 return True
136 return False
139def _delete_all_images(ecr: Any, repository_name: str) -> int:
140 """BatchDeleteImage every image in the repository.
142 Returns the number of images deleted. ECR's ``batch_delete_image``
143 accepts up to 100 IDs per call so we paginate through both
144 ``describe_images`` (to discover digests) and the chunked delete.
145 """
146 digests: list[dict[str, str]] = []
147 paginator = ecr.get_paginator("describe_images")
148 for page in paginator.paginate(repositoryName=repository_name):
149 for detail in page.get("imageDetails", []):
150 digest = detail.get("imageDigest")
151 if digest:
152 digests.append({"imageDigest": digest})
154 deleted = 0
155 for chunk_start in range(0, len(digests), 100):
156 # range() stops before len(digests), so every slice holds at least
157 # one digest and at most 100 — ECR's BatchDeleteImage ceiling.
158 chunk = digests[chunk_start : chunk_start + 100]
159 resp = ecr.batch_delete_image(
160 repositoryName=repository_name,
161 imageIds=chunk,
162 )
163 deleted += len(resp.get("imageIds", []))
164 return deleted
167def _handle_create_or_update(ecr: Any, properties: dict[str, Any]) -> dict[str, Any]:
168 """Adopt-or-create the repository and apply the lifecycle policy."""
169 repository_name = properties["RepositoryName"]
170 lifecycle_policy = properties.get("LifecyclePolicy")
172 existing = _describe_repository(ecr, repository_name)
173 if existing is not None:
174 repository_arn = existing.get("repositoryArn")
175 repository_uri = existing.get("repositoryUri")
176 adopted = True
177 else:
178 created = _create_repository(ecr, repository_name)
179 repository_arn = created.get("repositoryArn")
180 repository_uri = created.get("repositoryUri")
181 adopted = False
183 if lifecycle_policy:
184 _apply_lifecycle_policy(ecr, repository_name, lifecycle_policy)
186 return {
187 "PhysicalResourceId": repository_arn or repository_name,
188 "Data": {
189 "RepositoryArn": repository_arn or "",
190 "RepositoryUri": repository_uri or "",
191 "RepositoryName": repository_name,
192 "Adopted": "true" if adopted else "false",
193 },
194 }
197def _handle_delete(ecr: Any, properties: dict[str, Any], physical_id: str) -> dict[str, Any]:
198 """Honor the retain tag, removal policy, and empty-on-delete switches."""
199 repository_name = properties["RepositoryName"]
200 removal_policy = str(properties.get("RemovalPolicy", "retain")).lower()
201 empty_on_delete = bool(properties.get("EmptyOnDelete", False))
203 existing = _describe_repository(ecr, repository_name)
204 if existing is None:
205 # Already gone — treat as success.
206 logger.info("Repository %s already absent on Delete; skipping.", repository_name)
207 return {"PhysicalResourceId": physical_id, "Data": {"Deleted": "false"}}
209 repository_arn = existing.get("repositoryArn", physical_id)
211 if removal_policy != "destroy":
212 logger.info(
213 "removal_policy=%s for %s; leaving the repository in place.",
214 removal_policy,
215 repository_name,
216 )
217 return {
218 "PhysicalResourceId": physical_id,
219 "Data": {"Deleted": "false", "Reason": "removal-policy-retain"},
220 }
222 if _has_retain_tag(ecr, repository_arn):
223 logger.info(
224 "Repository %s carries gco:retain=true; preserving despite removal_policy=%s.",
225 repository_name,
226 removal_policy,
227 )
228 return {
229 "PhysicalResourceId": physical_id,
230 "Data": {"Deleted": "false", "Reason": "retain-tag"},
231 }
233 if empty_on_delete:
234 deleted = _delete_all_images(ecr, repository_name)
235 logger.info("Deleted %d images from %s before repo deletion.", deleted, repository_name)
237 ecr.delete_repository(repositoryName=repository_name, force=False)
238 return {"PhysicalResourceId": physical_id, "Data": {"Deleted": "true"}}
241def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
242 """Lookup-or-create custom resource handler for ECR repositories.
244 Dispatches on ``event["RequestType"]`` and returns the standard
245 CloudFormation custom resource response shape. The CDK Provider
246 framework wraps this dict into the protocol-required envelope.
247 """
248 request_type = event.get("RequestType", "")
249 properties = event.get("ResourceProperties", {}) or {}
250 physical_id = event.get("PhysicalResourceId", "")
251 logger.info("Image-Lookup CR event: %s for %s", request_type, properties.get("RepositoryName"))
253 ecr = _ecr_client()
255 if request_type in ("Create", "Update"):
256 return _handle_create_or_update(ecr, properties)
257 if request_type == "Delete":
258 return _handle_delete(ecr, properties, physical_id)
260 raise ValueError(f"Unsupported RequestType: {request_type!r}")