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