Coverage for cli/commands/images_cmd.py: 93.02%
341 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Container image registry commands.
3Subcommands wrap :class:`cli.images.ImageManager`. Read-only commands
4(`list`, `tags`, `describe`, `uri`, replication get/status) need no
5confirmation; administrative commands (`init`, `lifecycle`, replication
6sync) are idempotent; destructive commands (`delete-tag`, `delete-repo`,
7`cleanup`, `prune`) require ``-y`` / ``--yes``.
8"""
10from __future__ import annotations
12import json
13import sys
14from typing import Any
16import click
18from ..config import GCOConfig
19from ..output import get_output_formatter
21pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
24def _discard_mirror_log(_message: str) -> None:
25 """Suppress human progress logs when machine-readable output is requested."""
28@click.group()
29@pass_config
30def images(config: Any) -> None:
31 """Manage container images in the project ECR registry (gco/* repos)."""
32 pass
35# ---------------------------------------------------------------------------
36# Administrative
37# ---------------------------------------------------------------------------
40@images.command("init")
41@click.argument("name")
42@click.option("--retain/--no-retain", default=False, help="Apply gco:retain=true tag")
43@pass_config
44def images_init(config: Any, name: Any, retain: Any) -> None:
45 """Create a project repository with the default lifecycle policy.
47 Examples:
48 gco images init my-app
49 gco images init my-app --retain
50 """
51 from ..images import get_image_manager
53 formatter = get_output_formatter(config)
54 try:
55 manager = get_image_manager(config)
56 result = manager.init(name, retain=retain)
57 if config.output_format == "table":
58 if result.get("created"):
59 formatter.print_success(f"Created repository {result['name']}")
60 else:
61 formatter.print_info(f"Repository {result['name']} already existed")
62 else:
63 formatter.print(result)
64 except Exception as e:
65 formatter.print_error(f"Failed to init repository: {e}")
66 sys.exit(1)
69# ---------------------------------------------------------------------------
70# Read-only
71# ---------------------------------------------------------------------------
74@images.command("list")
75@pass_config
76def images_list(config: Any) -> None:
77 """List every repository under the project's gco/ prefix."""
78 from ..images import get_image_manager
80 formatter = get_output_formatter(config)
81 try:
82 repos = get_image_manager(config).list_repos()
83 if not repos:
84 formatter.print_info("No repositories found.")
85 return
86 formatter.print(repos)
87 except Exception as e:
88 formatter.print_error(f"Failed to list repositories: {e}")
89 sys.exit(1)
92@images.command("tags")
93@click.argument("name")
94@pass_config
95def images_tags(config: Any, name: Any) -> None:
96 """List tags within a repository."""
97 from ..images import get_image_manager
99 formatter = get_output_formatter(config)
100 try:
101 rows = get_image_manager(config).list_tags(name)
102 if not rows:
103 formatter.print_info("No tags found.")
104 return
105 formatter.print(rows)
106 except Exception as e:
107 formatter.print_error(f"Failed to list tags: {e}")
108 sys.exit(1)
111@images.command("describe")
112@click.argument("name")
113@click.argument("tag")
114@pass_config
115def images_describe(config: Any, name: Any, tag: Any) -> None:
116 """Print the full ECR details for a single image tag."""
117 from ..images import get_image_manager
119 formatter = get_output_formatter(config)
120 try:
121 result = get_image_manager(config).describe(name, tag)
122 if not result:
123 formatter.print_info(f"Tag '{tag}' not found in {name}")
124 return
125 formatter.print(result)
126 except Exception as e:
127 formatter.print_error(f"Failed to describe image: {e}")
128 sys.exit(1)
131@images.command("uri")
132@click.argument("name")
133@click.option("--tag", "-t", default="latest", help="Image tag (default: latest)")
134@pass_config
135def images_uri(config: Any, name: Any, tag: Any) -> None:
136 """Print the registry URI for an image without making any AWS calls."""
137 from ..images import get_image_manager
139 formatter = get_output_formatter(config)
140 try:
141 uri = get_image_manager(config).get_uri(name, tag=tag)
142 print(uri)
143 except Exception as e:
144 formatter.print_error(f"Failed to compute URI: {e}")
145 sys.exit(1)
148# ---------------------------------------------------------------------------
149# Build / push
150# ---------------------------------------------------------------------------
153@images.command("build")
154@click.argument("context")
155@click.option("--name", "-n", required=True, help="Image name")
156@click.option("--tag", "-t", default=None, help="Image tag (default: git SHA or 'latest')")
157@click.option("--dockerfile", "-f", default="Dockerfile", help="Path to Dockerfile")
158@click.option("--build-arg", "build_args", multiple=True, help="Build arg KEY=VALUE")
159@click.option("--platform", default="linux/amd64", help="Target platform")
160@click.option("--retain/--no-retain", default=False, help="Apply gco:retain=true tag")
161@pass_config
162def images_build(
163 config: Any,
164 context: Any,
165 name: Any,
166 tag: Any,
167 dockerfile: Any,
168 build_args: Any,
169 platform: Any,
170 retain: Any,
171) -> None:
172 """Build a container image and push it to the project's ECR repo.
174 Examples:
175 gco images build ./my-app --name my-app --tag v1
176 gco images build ./svc --name svc --build-arg VERSION=1.2.3
177 """
178 from ..images import get_image_manager
180 formatter = get_output_formatter(config)
182 args_dict: dict[str, str] = {}
183 for arg in build_args or ():
184 if "=" not in arg:
185 formatter.print_error(f"Invalid --build-arg (missing '='): {arg}")
186 sys.exit(1)
187 key, value = arg.split("=", 1)
188 args_dict[key] = value
190 try:
191 manager = get_image_manager(config)
192 result = manager.build(
193 context=context,
194 name=name,
195 tag=tag,
196 dockerfile=dockerfile,
197 build_args=args_dict or None,
198 platform=platform,
199 retain=retain,
200 quiet=config.output_format != "table",
201 )
202 if config.output_format == "table":
203 formatter.print_success(f"Built and pushed {result['image_uri']}")
204 if result.get("digest"): 204 ↛ exitline 204 didn't return from function 'images_build' because the condition on line 204 was always true
205 formatter.print_info(f"Digest: {result['digest']}")
206 else:
207 formatter.print(result)
208 except Exception as e:
209 formatter.print_error(f"Failed to build image: {e}")
210 sys.exit(1)
213@images.command("push")
214@click.argument("name")
215@click.option("--tag", "-t", required=True, help="Image tag")
216@click.option("--local-image", required=True, help="Existing local image reference")
217@click.option("--retain/--no-retain", default=False, help="Apply gco:retain=true tag")
218@pass_config
219def images_push(
220 config: Any,
221 name: Any,
222 tag: Any,
223 local_image: Any,
224 retain: Any,
225) -> None:
226 """Push an already-built local image to the project's ECR repo."""
227 from ..images import get_image_manager
229 formatter = get_output_formatter(config)
230 try:
231 result = get_image_manager(config).push(
232 name=name,
233 tag=tag,
234 local_image=local_image,
235 retain=retain,
236 quiet=config.output_format != "table",
237 )
238 if config.output_format == "table":
239 formatter.print_success(f"Pushed {result['image_uri']}")
240 if result.get("digest"): 240 ↛ exitline 240 didn't return from function 'images_push' because the condition on line 240 was always true
241 formatter.print_info(f"Digest: {result['digest']}")
242 else:
243 formatter.print(result)
244 except Exception as e:
245 formatter.print_error(f"Failed to push image: {e}")
246 sys.exit(1)
249# ---------------------------------------------------------------------------
250# Mirror third-party images into the project ECR
251# ---------------------------------------------------------------------------
254@images.command("mirror")
255@click.option(
256 "--region",
257 "-r",
258 required=True,
259 help="Target AWS region (must match the regional stack, e.g. us-east-1).",
260)
261@click.option(
262 "--ecr-namespace",
263 default=None,
264 help=(
265 "Destination ECR namespace. Defaults to cdk.json "
266 "volcano_image_mirror.ecr_namespace (gco/dockerhub); must match it so the "
267 "consumer's image override resolves to the mirror."
268 ),
269)
270@click.option(
271 "--no-skip-existing",
272 is_flag=True,
273 default=False,
274 help="Re-copy images even if the tag already exists in ECR.",
275)
276@click.option(
277 "--dry-run",
278 is_flag=True,
279 default=False,
280 help="Print the copy plan without creating repositories or copying images.",
281)
282@pass_config
283def images_mirror(
284 config: Any,
285 region: Any,
286 ecr_namespace: Any,
287 no_skip_existing: Any,
288 dry_run: Any,
289) -> None:
290 """Mirror third-party images (e.g. Volcano's docker.io images) into the ECR.
292 This is the same multi-arch copy ``gco stacks deploy`` runs automatically when
293 ``volcano_image_mirror.enabled`` is set. Run it directly to pre-seed a region
294 before enabling the toggle, or to re-mirror after bumping a mirrored image's
295 version. Wraps the shared ``cli._image_mirror`` core (also used by the deploy
296 auto-mirror and the ``images_mirror`` MCP tool).
298 Examples:
299 gco images mirror --region us-east-1
300 gco images mirror --region us-east-1 --dry-run
301 gco images mirror --region us-east-1 --ecr-namespace gco/dockerhub
302 """
303 from .. import _image_mirror as mirror
305 formatter = get_output_formatter(config)
306 namespace = (ecr_namespace or mirror.cdk_default_namespace()).strip("/")
307 table_output = config.output_format == "table"
308 try:
309 if dry_run:
310 # Partition metadata is local, so this remains free of AWS API calls.
311 registry_host = mirror._registry_host("<account>", region)
312 plan = mirror.plan_from_sources(mirror.collect_source_refs(), registry_host, namespace)
313 result = {
314 "region": region,
315 "ecr_namespace": namespace,
316 "images": [
317 {"source_ref": item.source_ref, "dest_ref": item.dest_ref} for item in plan
318 ],
319 }
320 if table_output:
321 formatter.print_info(
322 f"[dry-run] would mirror {len(plan)} image(s) into namespace {namespace!r}:"
323 )
324 for item in plan:
325 formatter.print_info(f" {item.source_ref} -> {item.dest_ref}")
326 else:
327 formatter.print(result)
328 return
330 result = mirror.mirror_images(
331 region,
332 ecr_namespace=namespace,
333 skip_existing=not no_skip_existing,
334 log=print if table_output else _discard_mirror_log,
335 )
336 if table_output: 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true
337 formatter.print_success(
338 f"Mirrored {len(result['mirrored'])}, skipped {len(result['skipped'])} "
339 f"into {result['registry']} (strategy: {result['strategy']})."
340 )
341 else:
342 formatter.print(result)
343 except Exception as e:
344 formatter.print_error(f"Failed to mirror images: {e}")
345 sys.exit(1)
348# ---------------------------------------------------------------------------
349# Destructive
350# ---------------------------------------------------------------------------
353@images.command("delete-tag")
354@click.argument("name")
355@click.argument("tag")
356@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
357@pass_config
358def images_delete_tag(config: Any, name: Any, tag: Any, yes: Any) -> None:
359 """Delete a single tag from a repository (irreversible)."""
360 from ..images import get_image_manager
362 formatter = get_output_formatter(config)
363 if not yes: 363 ↛ 364line 363 didn't jump to line 364 because the condition on line 363 was never true
364 formatter.print_error("--yes is required for destructive commands")
365 sys.exit(1)
366 try:
367 result = get_image_manager(config).delete_tag(name, tag)
368 formatter.print_success(
369 f"Deleted {len(result.get('deleted', []))} image(s) from {result['name']}"
370 )
371 if config.output_format != "table": 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 formatter.print(result)
373 except Exception as e:
374 formatter.print_error(f"Failed to delete tag: {e}")
375 sys.exit(1)
378@images.command("delete-repo")
379@click.argument("name")
380@click.option("--force/--no-force", default=False, help="Delete even if non-empty")
381@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
382@pass_config
383def images_delete_repo(config: Any, name: Any, force: Any, yes: Any) -> None:
384 """Delete a whole repository (irreversible)."""
385 from ..images import get_image_manager
387 formatter = get_output_formatter(config)
388 if not yes: 388 ↛ 389line 388 didn't jump to line 389 because the condition on line 388 was never true
389 formatter.print_error("--yes is required for destructive commands")
390 sys.exit(1)
391 try:
392 result = get_image_manager(config).delete_repo(name, force=force)
393 formatter.print_success(f"Deleted repository {result['name']}")
394 if config.output_format != "table": 394 ↛ 395line 394 didn't jump to line 395 because the condition on line 394 was never true
395 formatter.print(result)
396 except Exception as e:
397 formatter.print_error(f"Failed to delete repository: {e}")
398 sys.exit(1)
401@images.command("cleanup")
402@click.option("--name", "-n", default=None, help="Single repository to clean up")
403@click.option("--all", "all_repos", is_flag=True, help="Clean up every project repo")
404@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
405@pass_config
406def images_cleanup(config: Any, name: Any, all_repos: Any, yes: Any) -> None:
407 """Remove untagged images across one or all project repos."""
408 from ..images import get_image_manager
410 formatter = get_output_formatter(config)
411 if not yes: 411 ↛ 412line 411 didn't jump to line 412 because the condition on line 411 was never true
412 formatter.print_error("--yes is required for destructive commands")
413 sys.exit(1)
414 if not name and not all_repos:
415 formatter.print_error("Provide --name <repo> or --all")
416 sys.exit(1)
417 try:
418 result = get_image_manager(config).cleanup(name=name, all=all_repos)
419 formatter.print_success(
420 f"Cleaned up: repos_touched={result['repos_touched']} "
421 f"tags_deleted={result['tags_deleted']} "
422 f"bytes_freed={result['bytes_freed']}"
423 )
424 if config.output_format != "table": 424 ↛ 425line 424 didn't jump to line 425 because the condition on line 424 was never true
425 formatter.print(result)
426 except Exception as e:
427 formatter.print_error(f"Failed to clean up: {e}")
428 sys.exit(1)
431@images.command("prune")
432@click.option(
433 "--dry-run/--no-dry-run",
434 default=True,
435 help="Dry run by default; pass --no-dry-run to actually delete",
436)
437@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
438@pass_config
439def images_prune(config: Any, dry_run: Any, yes: Any) -> None:
440 """Remove untagged images older than 30 days (dry-run by default)."""
441 from ..images import get_image_manager
443 formatter = get_output_formatter(config)
444 if not yes: 444 ↛ 445line 444 didn't jump to line 445 because the condition on line 444 was never true
445 formatter.print_error("--yes is required for destructive commands")
446 sys.exit(1)
447 try:
448 result = get_image_manager(config).prune(dry_run=dry_run)
449 verb = "Would delete" if dry_run else "Deleted"
450 formatter.print_success(
451 f"{verb}: repos_touched={result['repos_touched']} "
452 f"tags_deleted={result['tags_deleted']} "
453 f"bytes_freed={result['bytes_freed']}"
454 )
455 if config.output_format != "table": 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 formatter.print(result)
457 except Exception as e:
458 formatter.print_error(f"Failed to prune: {e}")
459 sys.exit(1)
462@images.command("orphans")
463@click.option(
464 "--threshold-days",
465 default=30,
466 type=int,
467 help="Only report tags older than this many days",
468)
469@pass_config
470def images_orphans(config: Any, threshold_days: Any) -> None:
471 """List tags older than threshold_days that are not referenced anywhere."""
472 from ..images import get_image_manager
474 formatter = get_output_formatter(config)
475 try:
476 rows = get_image_manager(config).orphans(threshold_days=threshold_days)
477 if not rows:
478 formatter.print_info("No orphans found.")
479 return
480 formatter.print(rows)
481 except Exception as e:
482 formatter.print_error(f"Failed to detect orphans: {e}")
483 sys.exit(1)
486# ---------------------------------------------------------------------------
487# Lifecycle
488# ---------------------------------------------------------------------------
491@images.group("lifecycle")
492def lifecycle() -> None:
493 """Lifecycle policy management."""
494 pass
497@lifecycle.command("get")
498@click.argument("name")
499@pass_config
500def lifecycle_get(config: Any, name: Any) -> None:
501 """Print the lifecycle policy on a repository."""
502 from ..images import get_image_manager
504 formatter = get_output_formatter(config)
505 try:
506 result = get_image_manager(config).lifecycle_get(name)
507 if not result:
508 formatter.print_info(f"No lifecycle policy on {name}.")
509 return
510 formatter.print(result)
511 except Exception as e:
512 formatter.print_error(f"Failed to read lifecycle policy: {e}")
513 sys.exit(1)
516@lifecycle.command("set")
517@click.argument("name")
518@click.option("--file", "-f", "policy_file", required=True, help="Path to lifecycle JSON")
519@pass_config
520def lifecycle_set(config: Any, name: Any, policy_file: Any) -> None:
521 """Replace the lifecycle policy on a repository from a JSON file."""
522 from ..images import get_image_manager
524 formatter = get_output_formatter(config)
525 try:
526 with open(policy_file, encoding="utf-8") as f:
527 policy = json.load(f)
528 result = get_image_manager(config).lifecycle_set(name, policy)
529 formatter.print_success(f"Updated lifecycle policy on {result['name']}")
530 if config.output_format != "table": 530 ↛ 531line 530 didn't jump to line 531 because the condition on line 530 was never true
531 formatter.print(result)
532 except Exception as e:
533 formatter.print_error(f"Failed to set lifecycle policy: {e}")
534 sys.exit(1)
537# ---------------------------------------------------------------------------
538# Replication
539# ---------------------------------------------------------------------------
542@images.group("replication")
543def replication() -> None:
544 """Replication management."""
545 pass
548@replication.command("get")
549@pass_config
550def replication_get(config: Any) -> None:
551 """Print the current ECR replication configuration."""
552 from ..images import get_image_manager
554 formatter = get_output_formatter(config)
555 try:
556 result = get_image_manager(config).replication_get()
557 if not result:
558 formatter.print_info("No replication policy configured.")
559 return
560 formatter.print(result)
561 except Exception as e:
562 formatter.print_error(f"Failed to read replication policy: {e}")
563 sys.exit(1)
566@replication.command("status")
567@pass_config
568def replication_status(config: Any) -> None:
569 """Print per-image replication status across project repos."""
570 from ..images import get_image_manager
572 formatter = get_output_formatter(config)
573 try:
574 rows = get_image_manager(config).replication_status()
575 if not rows:
576 formatter.print_info("No replication status entries.")
577 return
578 formatter.print(rows)
579 except Exception as e:
580 formatter.print_error(f"Failed to read replication status: {e}")
581 sys.exit(1)
584@replication.command("sync")
585@pass_config
586def replication_sync(config: Any) -> None:
587 """Apply the project's standard replication rule (gco/* to all regions)."""
588 from ..images import get_image_manager
590 formatter = get_output_formatter(config)
591 try:
592 result = get_image_manager(config).replication_sync()
593 dests = result.get("destinations") or []
594 formatter.print_success(
595 f"Replication rule synced: destinations={', '.join(dests) or 'none'}"
596 )
597 if config.output_format != "table": 597 ↛ 598line 597 didn't jump to line 598 because the condition on line 597 was never true
598 formatter.print(result)
599 except Exception as e:
600 formatter.print_error(f"Failed to sync replication rule: {e}")
601 sys.exit(1)