Coverage for cli / commands / images_cmd.py: 100.00%
329 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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"):
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"):
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:
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 try:
364 result = get_image_manager(config).delete_tag(name, tag)
365 formatter.print_success(
366 f"Deleted {len(result.get('deleted', []))} image(s) from {result['name']}"
367 )
368 if config.output_format != "table":
369 formatter.print(result)
370 except Exception as e:
371 formatter.print_error(f"Failed to delete tag: {e}")
372 sys.exit(1)
375@images.command("delete-repo")
376@click.argument("name")
377@click.option("--force/--no-force", default=False, help="Delete even if non-empty")
378@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
379@pass_config
380def images_delete_repo(config: Any, name: Any, force: Any, yes: Any) -> None:
381 """Delete a whole repository (irreversible)."""
382 from ..images import get_image_manager
384 formatter = get_output_formatter(config)
385 try:
386 result = get_image_manager(config).delete_repo(name, force=force)
387 formatter.print_success(f"Deleted repository {result['name']}")
388 if config.output_format != "table":
389 formatter.print(result)
390 except Exception as e:
391 formatter.print_error(f"Failed to delete repository: {e}")
392 sys.exit(1)
395@images.command("cleanup")
396@click.option("--name", "-n", default=None, help="Single repository to clean up")
397@click.option("--all", "all_repos", is_flag=True, help="Clean up every project repo")
398@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
399@pass_config
400def images_cleanup(config: Any, name: Any, all_repos: Any, yes: Any) -> None:
401 """Remove untagged images across one or all project repos."""
402 from ..images import get_image_manager
404 formatter = get_output_formatter(config)
405 if not name and not all_repos:
406 formatter.print_error("Provide --name <repo> or --all")
407 sys.exit(1)
408 try:
409 result = get_image_manager(config).cleanup(name=name, all=all_repos)
410 formatter.print_success(
411 f"Cleaned up: repos_touched={result['repos_touched']} "
412 f"tags_deleted={result['tags_deleted']} "
413 f"bytes_freed={result['bytes_freed']}"
414 )
415 if config.output_format != "table":
416 formatter.print(result)
417 except Exception as e:
418 formatter.print_error(f"Failed to clean up: {e}")
419 sys.exit(1)
422@images.command("prune")
423@click.option(
424 "--dry-run/--no-dry-run",
425 default=True,
426 help="Dry run by default; pass --no-dry-run to actually delete",
427)
428@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
429@pass_config
430def images_prune(config: Any, dry_run: Any, yes: Any) -> None:
431 """Remove untagged images older than 30 days (dry-run by default)."""
432 from ..images import get_image_manager
434 formatter = get_output_formatter(config)
435 try:
436 result = get_image_manager(config).prune(dry_run=dry_run)
437 verb = "Would delete" if dry_run else "Deleted"
438 formatter.print_success(
439 f"{verb}: repos_touched={result['repos_touched']} "
440 f"tags_deleted={result['tags_deleted']} "
441 f"bytes_freed={result['bytes_freed']}"
442 )
443 if config.output_format != "table":
444 formatter.print(result)
445 except Exception as e:
446 formatter.print_error(f"Failed to prune: {e}")
447 sys.exit(1)
450@images.command("orphans")
451@click.option(
452 "--threshold-days",
453 default=30,
454 type=int,
455 help="Only report tags older than this many days",
456)
457@pass_config
458def images_orphans(config: Any, threshold_days: Any) -> None:
459 """List tags older than threshold_days that are not referenced anywhere."""
460 from ..images import get_image_manager
462 formatter = get_output_formatter(config)
463 try:
464 rows = get_image_manager(config).orphans(threshold_days=threshold_days)
465 if not rows:
466 formatter.print_info("No orphans found.")
467 return
468 formatter.print(rows)
469 except Exception as e:
470 formatter.print_error(f"Failed to detect orphans: {e}")
471 sys.exit(1)
474# ---------------------------------------------------------------------------
475# Lifecycle
476# ---------------------------------------------------------------------------
479@images.group("lifecycle")
480def lifecycle() -> None:
481 """Lifecycle policy management."""
482 pass
485@lifecycle.command("get")
486@click.argument("name")
487@pass_config
488def lifecycle_get(config: Any, name: Any) -> None:
489 """Print the lifecycle policy on a repository."""
490 from ..images import get_image_manager
492 formatter = get_output_formatter(config)
493 try:
494 result = get_image_manager(config).lifecycle_get(name)
495 if not result:
496 formatter.print_info(f"No lifecycle policy on {name}.")
497 return
498 formatter.print(result)
499 except Exception as e:
500 formatter.print_error(f"Failed to read lifecycle policy: {e}")
501 sys.exit(1)
504@lifecycle.command("set")
505@click.argument("name")
506@click.option("--file", "-f", "policy_file", required=True, help="Path to lifecycle JSON")
507@pass_config
508def lifecycle_set(config: Any, name: Any, policy_file: Any) -> None:
509 """Replace the lifecycle policy on a repository from a JSON file."""
510 from ..images import get_image_manager
512 formatter = get_output_formatter(config)
513 try:
514 with open(policy_file, encoding="utf-8") as f:
515 policy = json.load(f)
516 result = get_image_manager(config).lifecycle_set(name, policy)
517 formatter.print_success(f"Updated lifecycle policy on {result['name']}")
518 if config.output_format != "table":
519 formatter.print(result)
520 except Exception as e:
521 formatter.print_error(f"Failed to set lifecycle policy: {e}")
522 sys.exit(1)
525# ---------------------------------------------------------------------------
526# Replication
527# ---------------------------------------------------------------------------
530@images.group("replication")
531def replication() -> None:
532 """Replication management."""
533 pass
536@replication.command("get")
537@pass_config
538def replication_get(config: Any) -> None:
539 """Print the current ECR replication configuration."""
540 from ..images import get_image_manager
542 formatter = get_output_formatter(config)
543 try:
544 result = get_image_manager(config).replication_get()
545 if not result:
546 formatter.print_info("No replication policy configured.")
547 return
548 formatter.print(result)
549 except Exception as e:
550 formatter.print_error(f"Failed to read replication policy: {e}")
551 sys.exit(1)
554@replication.command("status")
555@pass_config
556def replication_status(config: Any) -> None:
557 """Print per-image replication status across project repos."""
558 from ..images import get_image_manager
560 formatter = get_output_formatter(config)
561 try:
562 rows = get_image_manager(config).replication_status()
563 if not rows:
564 formatter.print_info("No replication status entries.")
565 return
566 formatter.print(rows)
567 except Exception as e:
568 formatter.print_error(f"Failed to read replication status: {e}")
569 sys.exit(1)
572@replication.command("sync")
573@pass_config
574def replication_sync(config: Any) -> None:
575 """Apply the project's standard replication rule (gco/* to all regions)."""
576 from ..images import get_image_manager
578 formatter = get_output_formatter(config)
579 try:
580 result = get_image_manager(config).replication_sync()
581 dests = result.get("destinations") or []
582 formatter.print_success(
583 f"Replication rule synced: destinations={', '.join(dests) or 'none'}"
584 )
585 if config.output_format != "table":
586 formatter.print(result)
587 except Exception as e:
588 formatter.print_error(f"Failed to sync replication rule: {e}")
589 sys.exit(1)