Coverage for cli/commands/images_cmd.py: 91%
311 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-15 15: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)
24@click.group()
25@pass_config
26def images(config: Any) -> None:
27 """Manage container images in the project ECR registry (gco/* repos)."""
28 pass
31# ---------------------------------------------------------------------------
32# Administrative
33# ---------------------------------------------------------------------------
36@images.command("init")
37@click.argument("name")
38@click.option("--retain/--no-retain", default=False, help="Apply gco:retain=true tag")
39@pass_config
40def images_init(config: Any, name: Any, retain: Any) -> None:
41 """Create a project repository with the default lifecycle policy.
43 Examples:
44 gco images init my-app
45 gco images init my-app --retain
46 """
47 from ..images import get_image_manager
49 formatter = get_output_formatter(config)
50 try:
51 manager = get_image_manager(config)
52 result = manager.init(name, retain=retain)
53 if result.get("created"):
54 formatter.print_success(f"Created repository {result['name']}")
55 else:
56 formatter.print_info(f"Repository {result['name']} already existed")
57 if config.output_format != "table": 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true
58 formatter.print(result)
59 except Exception as e:
60 formatter.print_error(f"Failed to init repository: {e}")
61 sys.exit(1)
64# ---------------------------------------------------------------------------
65# Read-only
66# ---------------------------------------------------------------------------
69@images.command("list")
70@pass_config
71def images_list(config: Any) -> None:
72 """List every repository under the project's gco/ prefix."""
73 from ..images import get_image_manager
75 formatter = get_output_formatter(config)
76 try:
77 repos = get_image_manager(config).list_repos()
78 if not repos:
79 formatter.print_info("No repositories found.")
80 return
81 formatter.print(repos)
82 except Exception as e:
83 formatter.print_error(f"Failed to list repositories: {e}")
84 sys.exit(1)
87@images.command("tags")
88@click.argument("name")
89@pass_config
90def images_tags(config: Any, name: Any) -> None:
91 """List tags within a repository."""
92 from ..images import get_image_manager
94 formatter = get_output_formatter(config)
95 try:
96 rows = get_image_manager(config).list_tags(name)
97 if not rows:
98 formatter.print_info("No tags found.")
99 return
100 formatter.print(rows)
101 except Exception as e:
102 formatter.print_error(f"Failed to list tags: {e}")
103 sys.exit(1)
106@images.command("describe")
107@click.argument("name")
108@click.argument("tag")
109@pass_config
110def images_describe(config: Any, name: Any, tag: Any) -> None:
111 """Print the full ECR details for a single image tag."""
112 from ..images import get_image_manager
114 formatter = get_output_formatter(config)
115 try:
116 result = get_image_manager(config).describe(name, tag)
117 if not result:
118 formatter.print_info(f"Tag '{tag}' not found in {name}")
119 return
120 formatter.print(result)
121 except Exception as e:
122 formatter.print_error(f"Failed to describe image: {e}")
123 sys.exit(1)
126@images.command("uri")
127@click.argument("name")
128@click.option("--tag", "-t", default="latest", help="Image tag (default: latest)")
129@pass_config
130def images_uri(config: Any, name: Any, tag: Any) -> None:
131 """Print the registry URI for an image without making any AWS calls."""
132 from ..images import get_image_manager
134 formatter = get_output_formatter(config)
135 try:
136 uri = get_image_manager(config).get_uri(name, tag=tag)
137 print(uri)
138 except Exception as e:
139 formatter.print_error(f"Failed to compute URI: {e}")
140 sys.exit(1)
143# ---------------------------------------------------------------------------
144# Build / push
145# ---------------------------------------------------------------------------
148@images.command("build")
149@click.argument("context")
150@click.option("--name", "-n", required=True, help="Image name")
151@click.option("--tag", "-t", default=None, help="Image tag (default: git SHA or 'latest')")
152@click.option("--dockerfile", "-f", default="Dockerfile", help="Path to Dockerfile")
153@click.option("--build-arg", "build_args", multiple=True, help="Build arg KEY=VALUE")
154@click.option("--platform", default="linux/amd64", help="Target platform")
155@click.option("--retain/--no-retain", default=False, help="Apply gco:retain=true tag")
156@pass_config
157def images_build(
158 config: Any,
159 context: Any,
160 name: Any,
161 tag: Any,
162 dockerfile: Any,
163 build_args: Any,
164 platform: Any,
165 retain: Any,
166) -> None:
167 """Build a container image and push it to the project's ECR repo.
169 Examples:
170 gco images build ./my-app --name my-app --tag v1
171 gco images build ./svc --name svc --build-arg VERSION=1.2.3
172 """
173 from ..images import get_image_manager
175 formatter = get_output_formatter(config)
177 args_dict: dict[str, str] = {}
178 for arg in build_args or ():
179 if "=" not in arg:
180 formatter.print_error(f"Invalid --build-arg (missing '='): {arg}")
181 sys.exit(1)
182 key, value = arg.split("=", 1)
183 args_dict[key] = value
185 try:
186 manager = get_image_manager(config)
187 result = manager.build(
188 context=context,
189 name=name,
190 tag=tag,
191 dockerfile=dockerfile,
192 build_args=args_dict or None,
193 platform=platform,
194 retain=retain,
195 )
196 formatter.print_success(f"Built and pushed {result['image_uri']}")
197 if result.get("digest"): 197 ↛ 199line 197 didn't jump to line 199 because the condition on line 197 was always true
198 formatter.print_info(f"Digest: {result['digest']}")
199 if config.output_format != "table": 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 formatter.print(result)
201 except Exception as e:
202 formatter.print_error(f"Failed to build image: {e}")
203 sys.exit(1)
206@images.command("push")
207@click.argument("name")
208@click.option("--tag", "-t", required=True, help="Image tag")
209@click.option("--local-image", required=True, help="Existing local image reference")
210@click.option("--retain/--no-retain", default=False, help="Apply gco:retain=true tag")
211@pass_config
212def images_push(
213 config: Any,
214 name: Any,
215 tag: Any,
216 local_image: Any,
217 retain: Any,
218) -> None:
219 """Push an already-built local image to the project's ECR repo."""
220 from ..images import get_image_manager
222 formatter = get_output_formatter(config)
223 try:
224 result = get_image_manager(config).push(
225 name=name, tag=tag, local_image=local_image, retain=retain
226 )
227 formatter.print_success(f"Pushed {result['image_uri']}")
228 if result.get("digest"): 228 ↛ 230line 228 didn't jump to line 230 because the condition on line 228 was always true
229 formatter.print_info(f"Digest: {result['digest']}")
230 if config.output_format != "table": 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true
231 formatter.print(result)
232 except Exception as e:
233 formatter.print_error(f"Failed to push image: {e}")
234 sys.exit(1)
237# ---------------------------------------------------------------------------
238# Destructive
239# ---------------------------------------------------------------------------
242@images.command("delete-tag")
243@click.argument("name")
244@click.argument("tag")
245@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
246@pass_config
247def images_delete_tag(config: Any, name: Any, tag: Any, yes: Any) -> None:
248 """Delete a single tag from a repository (irreversible)."""
249 from ..images import get_image_manager
251 formatter = get_output_formatter(config)
252 if not yes: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 formatter.print_error("--yes is required for destructive commands")
254 sys.exit(1)
255 try:
256 result = get_image_manager(config).delete_tag(name, tag)
257 formatter.print_success(
258 f"Deleted {len(result.get('deleted', []))} image(s) from {result['name']}"
259 )
260 if config.output_format != "table": 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 formatter.print(result)
262 except Exception as e:
263 formatter.print_error(f"Failed to delete tag: {e}")
264 sys.exit(1)
267@images.command("delete-repo")
268@click.argument("name")
269@click.option("--force/--no-force", default=False, help="Delete even if non-empty")
270@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
271@pass_config
272def images_delete_repo(config: Any, name: Any, force: Any, yes: Any) -> None:
273 """Delete a whole repository (irreversible)."""
274 from ..images import get_image_manager
276 formatter = get_output_formatter(config)
277 if not yes: 277 ↛ 278line 277 didn't jump to line 278 because the condition on line 277 was never true
278 formatter.print_error("--yes is required for destructive commands")
279 sys.exit(1)
280 try:
281 result = get_image_manager(config).delete_repo(name, force=force)
282 formatter.print_success(f"Deleted repository {result['name']}")
283 if config.output_format != "table": 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 formatter.print(result)
285 except Exception as e:
286 formatter.print_error(f"Failed to delete repository: {e}")
287 sys.exit(1)
290@images.command("cleanup")
291@click.option("--name", "-n", default=None, help="Single repository to clean up")
292@click.option("--all", "all_repos", is_flag=True, help="Clean up every project repo")
293@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
294@pass_config
295def images_cleanup(config: Any, name: Any, all_repos: Any, yes: Any) -> None:
296 """Remove untagged images across one or all project repos."""
297 from ..images import get_image_manager
299 formatter = get_output_formatter(config)
300 if not yes: 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 formatter.print_error("--yes is required for destructive commands")
302 sys.exit(1)
303 if not name and not all_repos:
304 formatter.print_error("Provide --name <repo> or --all")
305 sys.exit(1)
306 try:
307 result = get_image_manager(config).cleanup(name=name, all=all_repos)
308 formatter.print_success(
309 f"Cleaned up: repos_touched={result['repos_touched']} "
310 f"tags_deleted={result['tags_deleted']} "
311 f"bytes_freed={result['bytes_freed']}"
312 )
313 if config.output_format != "table": 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 formatter.print(result)
315 except Exception as e:
316 formatter.print_error(f"Failed to clean up: {e}")
317 sys.exit(1)
320@images.command("prune")
321@click.option(
322 "--dry-run/--no-dry-run",
323 default=True,
324 help="Dry run by default; pass --no-dry-run to actually delete",
325)
326@click.option("--yes", "-y", is_flag=True, required=True, help="Required confirmation")
327@pass_config
328def images_prune(config: Any, dry_run: Any, yes: Any) -> None:
329 """Remove untagged images older than 30 days (dry-run by default)."""
330 from ..images import get_image_manager
332 formatter = get_output_formatter(config)
333 if not yes: 333 ↛ 334line 333 didn't jump to line 334 because the condition on line 333 was never true
334 formatter.print_error("--yes is required for destructive commands")
335 sys.exit(1)
336 try:
337 result = get_image_manager(config).prune(dry_run=dry_run)
338 verb = "Would delete" if dry_run else "Deleted"
339 formatter.print_success(
340 f"{verb}: repos_touched={result['repos_touched']} "
341 f"tags_deleted={result['tags_deleted']} "
342 f"bytes_freed={result['bytes_freed']}"
343 )
344 if config.output_format != "table": 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 formatter.print(result)
346 except Exception as e:
347 formatter.print_error(f"Failed to prune: {e}")
348 sys.exit(1)
351@images.command("orphans")
352@click.option(
353 "--threshold-days",
354 default=30,
355 type=int,
356 help="Only report tags older than this many days",
357)
358@pass_config
359def images_orphans(config: Any, threshold_days: Any) -> None:
360 """List tags older than threshold_days that are not referenced anywhere."""
361 from ..images import get_image_manager
363 formatter = get_output_formatter(config)
364 try:
365 rows = get_image_manager(config).orphans(threshold_days=threshold_days)
366 if not rows:
367 formatter.print_info("No orphans found.")
368 return
369 formatter.print(rows)
370 except Exception as e:
371 formatter.print_error(f"Failed to detect orphans: {e}")
372 sys.exit(1)
375# ---------------------------------------------------------------------------
376# Lifecycle
377# ---------------------------------------------------------------------------
380@images.group("lifecycle")
381def lifecycle() -> None:
382 """Lifecycle policy management."""
383 pass
386@lifecycle.command("get")
387@click.argument("name")
388@pass_config
389def lifecycle_get(config: Any, name: Any) -> None:
390 """Print the lifecycle policy on a repository."""
391 from ..images import get_image_manager
393 formatter = get_output_formatter(config)
394 try:
395 result = get_image_manager(config).lifecycle_get(name)
396 if not result:
397 formatter.print_info(f"No lifecycle policy on {name}.")
398 return
399 formatter.print(result)
400 except Exception as e:
401 formatter.print_error(f"Failed to read lifecycle policy: {e}")
402 sys.exit(1)
405@lifecycle.command("set")
406@click.argument("name")
407@click.option("--file", "-f", "policy_file", required=True, help="Path to lifecycle JSON")
408@pass_config
409def lifecycle_set(config: Any, name: Any, policy_file: Any) -> None:
410 """Replace the lifecycle policy on a repository from a JSON file."""
411 from ..images import get_image_manager
413 formatter = get_output_formatter(config)
414 try:
415 with open(policy_file, encoding="utf-8") as f:
416 policy = json.load(f)
417 result = get_image_manager(config).lifecycle_set(name, policy)
418 formatter.print_success(f"Updated lifecycle policy on {result['name']}")
419 if config.output_format != "table": 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true
420 formatter.print(result)
421 except Exception as e:
422 formatter.print_error(f"Failed to set lifecycle policy: {e}")
423 sys.exit(1)
426# ---------------------------------------------------------------------------
427# Replication
428# ---------------------------------------------------------------------------
431@images.group("replication")
432def replication() -> None:
433 """Replication management."""
434 pass
437@replication.command("get")
438@pass_config
439def replication_get(config: Any) -> None:
440 """Print the current ECR replication configuration."""
441 from ..images import get_image_manager
443 formatter = get_output_formatter(config)
444 try:
445 result = get_image_manager(config).replication_get()
446 if not result:
447 formatter.print_info("No replication policy configured.")
448 return
449 formatter.print(result)
450 except Exception as e:
451 formatter.print_error(f"Failed to read replication policy: {e}")
452 sys.exit(1)
455@replication.command("status")
456@pass_config
457def replication_status(config: Any) -> None:
458 """Print per-image replication status across project repos."""
459 from ..images import get_image_manager
461 formatter = get_output_formatter(config)
462 try:
463 rows = get_image_manager(config).replication_status()
464 if not rows:
465 formatter.print_info("No replication status entries.")
466 return
467 formatter.print(rows)
468 except Exception as e:
469 formatter.print_error(f"Failed to read replication status: {e}")
470 sys.exit(1)
473@replication.command("sync")
474@pass_config
475def replication_sync(config: Any) -> None:
476 """Apply the project's standard replication rule (gco/* to all regions)."""
477 from ..images import get_image_manager
479 formatter = get_output_formatter(config)
480 try:
481 result = get_image_manager(config).replication_sync()
482 dests = result.get("destinations") or []
483 formatter.print_success(
484 f"Replication rule synced: destinations={', '.join(dests) or 'none'}"
485 )
486 if config.output_format != "table": 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true
487 formatter.print(result)
488 except Exception as e:
489 formatter.print_error(f"Failed to sync replication rule: {e}")
490 sys.exit(1)