1"""Plain Cloud CLI.
2
3Entrypoint for the `plain-cloud` binary. Subcommands authenticate against
4a Plain Cloud installation using a personal API key minted from the
5dashboard at /dashboard/api-keys/.
6"""
7
8import json
9import sys
10import webbrowser
11from pathlib import Path
12from typing import BinaryIO
13
14import click
15import httpx
16import keyring
17
18from .client import Client
19from .credentials import (
20 DEFAULT_API_URL,
21 Credentials,
22 KeyringUnavailable,
23 clear,
24 config_path,
25 load,
26 require,
27 save,
28)
29
30
31def _die(message: str, code: int = 1) -> None:
32 click.secho(message, fg="red", err=True)
33 sys.exit(code)
34
35
36def _write_passthrough(text: str) -> None:
37 if not text:
38 return
39 sys.stdout.write(text)
40 if not text.endswith("\n"):
41 sys.stdout.write("\n")
42
43
44def _split_kv(s: str, sep: str, flag: str) -> tuple[str, str]:
45 if sep not in s:
46 _die(f"{flag} expected KEY{sep}VALUE, got {s!r}")
47 key, _, value = s.partition(sep)
48 return key, value
49
50
51def _normalize_api_path(path: str) -> str:
52 # Client.base_url already mounts /api. Strip a redundant leading
53 # /api/ so users who copy from old docs (or the README's older
54 # examples) still hit the right URL.
55 stripped = path.lstrip("/")
56 if stripped == "api" or stripped.startswith("api/"):
57 stripped = stripped[3:].lstrip("/")
58 return "/" + stripped
59
60
61@click.group()
62@click.version_option(package_name="plain.cloud", prog_name="plain-cloud")
63def cli() -> None:
64 """Manage your Plain Cloud account from the command line."""
65
66
67@cli.command()
68@click.option(
69 "--api-url",
70 default=DEFAULT_API_URL,
71 show_default=True,
72 envvar="PLAIN_CLOUD_API_URL",
73 show_envvar=True,
74 help="Base URL of the Plain Cloud API.",
75)
76@click.option(
77 "--token",
78 default=None,
79 envvar="PLAIN_CLOUD_TOKEN",
80 show_envvar=True,
81 help="API token to save. If omitted, you'll be prompted.",
82)
83def login(api_url: str, token: str | None) -> None:
84 """Save an API token for future commands.
85
86 Mint a token at {api_url}/dashboard/api-keys/ and paste it here.
87 """
88 if token is None:
89 click.echo(
90 f"Mint a token at {api_url.rstrip('/')}/dashboard/api-keys/ and paste it below."
91 )
92 token = click.prompt("Token", hide_input=True).strip()
93 if not token:
94 _die("No token provided.")
95
96 creds = Credentials(api_url=api_url, token=token)
97 with Client(creds) as client:
98 try:
99 me = client.get("/me")
100 except httpx.HTTPError as exc:
101 _die(f"Could not reach {api_url}: {exc}")
102
103 if not isinstance(me, dict):
104 _die(
105 f"Unexpected response from {api_url} when verifying the token. "
106 "The token was not saved."
107 )
108
109 try:
110 backend = save(creds)
111 except KeyringUnavailable as exc:
112 _die(str(exc))
113
114 click.secho(
115 f"Logged in as {me.get('email') or me.get('username') or 'unknown'}.",
116 fg="green",
117 )
118 click.secho(f"Token stored in {backend}.", dim=True)
119
120
121@cli.command()
122def logout() -> None:
123 """Forget the saved API token."""
124 if clear():
125 click.secho("Logged out.", fg="green")
126 else:
127 click.secho("No saved credentials.", dim=True)
128
129
130@cli.command()
131def whoami() -> None:
132 """Show the user the current token belongs to."""
133 creds = require()
134 with Client(creds) as client:
135 me = client.get("/me")
136 if not isinstance(me, dict):
137 # require() already caught a missing token, so we have credentials —
138 # an empty/unexpected body here is a server-side problem, not logout.
139 _die(
140 f"Unexpected response from {creds.api_url} — could not read your identity.\n"
141 "If this persists, re-authenticate with `plain-cloud login`."
142 )
143 if email := me.get("email"):
144 identity = click.style(email, bold=True)
145 else:
146 identity = click.style("(no email)", dim=True)
147 click.echo(f"{identity} · {click.style(creds.api_url, dim=True)}")
148 if me.get("username"):
149 click.echo(f"username: {me['username']}")
150 teams = me.get("teams") or []
151 if teams:
152 click.echo(f"teams: {len(teams)}")
153 for team in teams:
154 click.echo(
155 f" - {click.style(team['slug'], fg='cyan')} "
156 f"{click.style('(' + team.get('role', '?') + ')', dim=True)}"
157 )
158 else:
159 click.secho("teams: (none)", dim=True)
160
161
162@cli.group()
163def apps() -> None:
164 """Manage Plain Cloud apps."""
165
166
167@apps.command("list")
168def apps_list() -> None:
169 """List apps you have access to."""
170 creds = require()
171 with Client(creds) as client:
172 data = client.get("/apps/")
173 rows = data.get("apps") or []
174 if not rows:
175 click.secho("No apps yet.", dim=True)
176 return
177 width = max(len(row["slug"]) for row in rows)
178 for row in rows:
179 slug = click.style(f"{row['slug']:<{width}}", fg="cyan")
180 team = click.style(row["team_slug"], dim=True)
181 click.echo(f" {slug} {team}")
182
183
184@cli.command("api")
185@click.argument("path")
186@click.option(
187 "-X",
188 "--method",
189 default="GET",
190 show_default=True,
191 type=click.Choice(["GET", "POST", "PUT", "PATCH", "DELETE"], case_sensitive=False),
192 help="HTTP method.",
193)
194@click.option(
195 "-H",
196 "--header",
197 "headers",
198 multiple=True,
199 metavar="KEY:VALUE",
200 help="Add a request header. Repeatable.",
201)
202@click.option(
203 "-f",
204 "--raw-field",
205 "raw_fields",
206 multiple=True,
207 metavar="KEY=VALUE",
208 help="Add a string field. Repeatable.",
209)
210@click.option(
211 "-F",
212 "--field",
213 "typed_fields",
214 multiple=True,
215 metavar="KEY=VALUE",
216 help=(
217 "Add a typed field (true/false/null and numbers become JSON literals; "
218 "@path reads a string from a file). Repeatable."
219 ),
220)
221@click.option(
222 "--input",
223 "input_source",
224 type=click.File("rb"),
225 default=None,
226 metavar="FILE",
227 help='Read raw request body from FILE. Use "-" for stdin.',
228)
229@click.option(
230 "--raw",
231 is_flag=True,
232 help="Print the response body verbatim, no JSON pretty-printing.",
233)
234def api(
235 path: str,
236 method: str,
237 headers: tuple[str, ...],
238 raw_fields: tuple[str, ...],
239 typed_fields: tuple[str, ...],
240 input_source: BinaryIO | None,
241 raw: bool,
242) -> None:
243 """Call any Plain Cloud API path with the saved token.
244
245 \b
246 PATH is the API path as listed in `plain-cloud openapi` (e.g. /apps/).
247 The /api/ prefix is added automatically; passing it explicitly also works.
248
249 \b
250 Fields go in the query string for GET requests and in the JSON body for
251 everything else. Use -f for plain strings, -F for typed values
252 (true/false/null/number, or @file to read a string from disk).
253
254 \b
255 Examples:
256 plain-cloud api /me/
257 plain-cloud api /apps/ -F page=2
258 plain-cloud api /apps/foo/exceptions/123/resolve/ -X POST
259 plain-cloud api /apps/ -X POST --input body.json -H "X-Trace: 1"
260
261 Exit code is 0 for 2xx, 1 for everything else.
262 """
263 creds = require()
264 method_upper = method.upper()
265 path = _normalize_api_path(path)
266
267 request_headers: dict[str, str] = {}
268 for header in headers:
269 key, value = _split_kv(header, ":", "--header")
270 request_headers[key.strip()] = value.lstrip()
271
272 fields: list[tuple[str, object]] = []
273 for raw_field in raw_fields:
274 key, value = _split_kv(raw_field, "=", "--raw-field")
275 fields.append((key, value))
276 for typed in typed_fields:
277 key, value = _split_kv(typed, "=", "--field")
278 fields.append((key, _coerce_field_value(value)))
279
280 body_content = input_source.read() if input_source is not None else None
281
282 request_kwargs: dict[str, object] = {}
283 if request_headers:
284 request_kwargs["headers"] = request_headers
285 if body_content is not None:
286 request_kwargs["content"] = body_content
287
288 if fields:
289 if body_content is None and method_upper != "GET":
290 request_kwargs["json"] = dict(fields)
291 else:
292 request_kwargs["params"] = [(k, _to_query_value(v)) for k, v in fields]
293
294 with Client(creds) as client:
295 try:
296 response = client.raw_request(method_upper, path, **request_kwargs)
297 except httpx.HTTPError as exc:
298 _die(f"Request failed: {exc}")
299
300 if raw or not response.content:
301 _write_passthrough(response.text)
302 else:
303 try:
304 payload = response.json()
305 except ValueError:
306 _write_passthrough(response.text)
307 else:
308 click.echo(json.dumps(payload, indent=2))
309
310 if response.status_code >= 400:
311 click.secho(
312 f"({method_upper} {path} → {response.status_code})",
313 fg="red",
314 err=True,
315 )
316 sys.exit(1)
317
318
319def _coerce_field_value(value: str) -> object:
320 # Mirrors `gh api -F` so users with gh muscle memory get the same coercion.
321 if value == "true":
322 return True
323 if value == "false":
324 return False
325 if value == "null":
326 return None
327 if value.startswith("@"):
328 try:
329 return Path(value[1:]).read_text().rstrip("\n")
330 except OSError as exc:
331 _die(f"--field could not read {value[1:]}: {exc}")
332 try:
333 return int(value)
334 except ValueError:
335 pass
336 try:
337 return float(value)
338 except ValueError:
339 pass
340 return value
341
342
343def _to_query_value(value: object) -> str:
344 # Lowercase true/false/null match `gh api`'s query rendering.
345 if value is True:
346 return "true"
347 if value is False:
348 return "false"
349 if value is None:
350 return "null"
351 return str(value)
352
353
354@cli.command("open")
355@click.argument("path", required=False, default="/dashboard/")
356@click.option(
357 "--api-url",
358 default=None,
359 envvar="PLAIN_CLOUD_API_URL",
360 show_envvar=True,
361 help="Override the api_url (useful when not logged in).",
362)
363def open_url(path: str, api_url: str | None) -> None:
364 """Open a Plain Cloud URL in your browser."""
365 if api_url is None:
366 creds = load()
367 api_url = creds.api_url if creds else DEFAULT_API_URL
368 base = api_url.rstrip("/")
369 if not path.startswith("/"):
370 path = "/" + path
371 url = base + path
372 click.secho(f"Opening {url}", dim=True)
373 webbrowser.open(url)
374
375
376@cli.command("openapi")
377@click.option(
378 "--raw",
379 is_flag=True,
380 help="Print the response body verbatim, no JSON pretty-printing.",
381)
382@click.option(
383 "--api-url",
384 default=None,
385 envvar="PLAIN_CLOUD_API_URL",
386 show_envvar=True,
387 help="Override the api_url (useful when not logged in).",
388)
389def openapi(raw: bool, api_url: str | None) -> None:
390 """Fetch the OpenAPI document for this Plain Cloud install.
391
392 The schema is metadata, so no token is required. When logged in, the
393 saved api_url is used; pass --api-url to point elsewhere or to fetch
394 without logging in.
395 """
396 if api_url is None:
397 creds = load()
398 api_url = creds.api_url if creds else DEFAULT_API_URL
399
400 url = api_url.rstrip("/") + "/api/openapi.json"
401 try:
402 response = httpx.get(url, timeout=httpx.Timeout(30.0))
403 except httpx.HTTPError as exc:
404 _die(f"Could not reach {url}: {exc}")
405
406 if response.status_code >= 400:
407 _die(f"{response.status_code} fetching {url}: {response.text[:200]}")
408
409 if raw:
410 _write_passthrough(response.text)
411 return
412 try:
413 click.echo(json.dumps(response.json(), indent=2, sort_keys=False))
414 except ValueError:
415 _write_passthrough(response.text)
416
417
418@cli.command()
419def config() -> None:
420 """Show where credentials are stored."""
421 creds = load()
422
423 def row(label: str, value: str) -> None:
424 click.echo(f"{click.style(label, bold=True)} {value}")
425
426 row("Config: ", str(config_path()))
427 row("Keyring:", keyring.get_keyring().name)
428 if creds:
429 row("API: ", creds.api_url)
430 row("Status: ", click.style("logged in", fg="green"))
431 else:
432 row("Status: ", click.style("not logged in", dim=True))
433
434
435if __name__ == "__main__":
436 cli()