Coverage for cli / commands / webhooks_cmd.py: 100.00%
101 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Webhook commands."""
3import sys
4from typing import Any
6import click
8from ..config import GCOConfig
9from ..output import confirm, get_output_formatter
11pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
14@click.group()
15@pass_config
16def webhooks(config: Any) -> None:
17 """Manage webhooks for job event notifications.
19 Webhooks receive HTTP POST notifications when job events occur
20 (job.started, job.completed, job.failed).
21 """
24@webhooks.command("list")
25@click.option("--namespace", "-n", help="Filter by namespace")
26@click.option("--region", "-r", help="Region to query (any region works)")
27@pass_config
28def webhooks_list(config: Any, namespace: Any, region: Any) -> None:
29 """List all registered webhooks.
31 Examples:
32 gco webhooks list
33 gco webhooks list --namespace gco-jobs
34 """
35 formatter = get_output_formatter(config)
37 try:
38 from ..aws_client import get_aws_client
40 aws_client = get_aws_client(config)
42 query_region = region or (config.default_region if config.use_regional_api else None)
43 params = {}
44 if namespace:
45 params["namespace"] = namespace
47 result = aws_client.call_api(
48 method="GET",
49 path="/api/v1/webhooks",
50 region=query_region,
51 params=params,
52 )
54 if config.output_format == "table":
55 webhooks_data = result.get("webhooks", [])
56 if not webhooks_data:
57 formatter.print_info("No webhooks found")
58 return
60 print(f"\n Webhooks ({result.get('count', 0)} total)")
61 print(" " + "-" * 80)
62 print(
63 " ID URL EVENTS NAMESPACE"
64 )
65 print(" " + "-" * 80)
66 for w in webhooks_data:
67 wid = w.get("id", "")[:8]
68 url = w.get("url", "")[:40]
69 events = ",".join(w.get("events", []))[:18]
70 ns = (w.get("namespace") or "all")[:12]
71 print(f" {wid:<9} {url:<42} {events:<19} {ns}")
72 else:
73 formatter.print(result)
75 except Exception as e:
76 formatter.print_error(f"Failed to list webhooks: {e}")
77 sys.exit(1)
80@webhooks.command("get")
81@click.argument("webhook_id")
82@click.option("--region", "-r", help="Region to query (any region works)")
83@pass_config
84def webhooks_get(config: Any, webhook_id: Any, region: Any) -> None:
85 """Get a single webhook by id.
87 The webhooks API exposes list/create/delete but no fetch-by-id endpoint, so
88 this lists the region's webhooks and returns the one whose id matches.
90 Examples:
91 gco webhooks get abc12345
92 gco webhooks get abc12345 -r us-east-1
93 """
94 formatter = get_output_formatter(config)
96 try:
97 from ..aws_client import get_aws_client
99 aws_client = get_aws_client(config)
101 query_region = region or (config.default_region if config.use_regional_api else None)
102 result = aws_client.call_api(
103 method="GET",
104 path="/api/v1/webhooks",
105 region=query_region,
106 )
107 match = next(
108 (w for w in result.get("webhooks", []) if w.get("id") == webhook_id),
109 None,
110 )
111 if match is None:
112 formatter.print_error(f"Webhook '{webhook_id}' not found")
113 sys.exit(1)
114 formatter.print(match)
116 except SystemExit:
117 raise
118 except Exception as e:
119 formatter.print_error(f"Failed to get webhook: {e}")
120 sys.exit(1)
123@webhooks.command("create")
124@click.option("--url", "-u", required=True, help="Webhook URL")
125@click.option(
126 "--event",
127 "-e",
128 multiple=True,
129 required=True,
130 type=click.Choice(["job.started", "job.completed", "job.failed"]),
131 help="Events to subscribe to",
132)
133@click.option("--namespace", "-n", help="Filter events by namespace")
134@click.option("--secret", "-s", help="Secret for HMAC signature verification")
135@click.option("--region", "-r", help="Region to use (any region works)")
136@pass_config
137def webhooks_create(
138 config: Any, url: Any, event: Any, namespace: Any, secret: Any, region: Any
139) -> None:
140 """Register a new webhook for job events.
142 Examples:
143 gco webhooks create --url https://example.com/webhook -e job.completed -e job.failed
144 gco webhooks create -u https://slack.com/webhook -e job.failed -n gco-jobs
145 """
146 formatter = get_output_formatter(config)
148 try:
149 from ..aws_client import get_aws_client
151 aws_client = get_aws_client(config)
153 query_region = region or (config.default_region if config.use_regional_api else None)
154 result = aws_client.call_api(
155 method="POST",
156 path="/api/v1/webhooks",
157 region=query_region,
158 body={
159 "url": url,
160 "events": list(event),
161 "namespace": namespace,
162 "secret": secret,
163 },
164 )
166 formatter.print_success("Webhook registered successfully")
167 formatter.print(result)
169 except Exception as e:
170 formatter.print_error(f"Failed to create webhook: {e}")
171 sys.exit(1)
174@webhooks.command("delete")
175@click.argument("webhook_id")
176@click.option("--region", "-r", help="Region to use (any region works)")
177@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
178@pass_config
179def webhooks_delete(config: Any, webhook_id: Any, region: Any, yes: Any) -> None:
180 """Delete a webhook.
182 Examples:
183 gco webhooks delete abc123
184 gco webhooks delete abc123 -y
185 """
186 formatter = get_output_formatter(config)
188 if not yes:
189 confirm(f"Delete webhook '{webhook_id}'?", abort=True)
191 try:
192 from ..aws_client import get_aws_client
194 aws_client = get_aws_client(config)
196 query_region = region or (config.default_region if config.use_regional_api else None)
197 result = aws_client.call_api(
198 method="DELETE",
199 path=f"/api/v1/webhooks/{webhook_id}",
200 region=query_region,
201 )
203 formatter.print_success(f"Webhook '{webhook_id}' deleted")
204 formatter.print(result)
206 except Exception as e:
207 formatter.print_error(f"Failed to delete webhook: {e}")
208 sys.exit(1)