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