This repository has no description
1"""Live knot git access for Tangled repos (tree + blob)."""
2
3from __future__ import annotations
4
5from typing import Any
6
7import httpx
8
9DEFAULT_TIMEOUT = httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=10.0)
10
11
12def knot_xrpc(
13 client: httpx.Client,
14 knot_hostname: str,
15 method: str,
16 params: dict[str, Any],
17) -> tuple[int, Any]:
18 host = knot_hostname.removeprefix("https://").rstrip("/")
19 resp = client.get(f"https://{host}/xrpc/{method}", params=params)
20 if resp.status_code != 200:
21 return resp.status_code, {"error": resp.status_code, "body": resp.text[:500]}
22 try:
23 return resp.status_code, resp.json()
24 except ValueError:
25 return resp.status_code, {"raw": resp.text[:500]}
26
27
28def list_tree(
29 client: httpx.Client,
30 *,
31 knot_hostname: str,
32 repo_did: str,
33 path: str = "",
34 ref: str = "HEAD",
35) -> dict[str, Any]:
36 status, payload = knot_xrpc(
37 client,
38 knot_hostname,
39 "sh.tangled.repo.tree",
40 {"repo": repo_did, "ref": ref, "path": path},
41 )
42 if status != 200 or not isinstance(payload, dict):
43 raise RuntimeError(f"tree failed HTTP {status}: {payload!r}")
44 return payload
45
46
47def read_blob(
48 client: httpx.Client,
49 *,
50 knot_hostname: str,
51 repo_did: str,
52 path: str,
53 ref: str = "HEAD",
54) -> str:
55 status, payload = knot_xrpc(
56 client,
57 knot_hostname,
58 "sh.tangled.repo.blob",
59 {"repo": repo_did, "ref": ref, "path": path},
60 )
61 if status != 200 or not isinstance(payload, dict):
62 raise RuntimeError(f"blob failed HTTP {status}: {payload!r}")
63 content = payload.get("content")
64 if not isinstance(content, str):
65 raise RuntimeError("blob response missing text content")
66 return content
67
68
69def describe_repo_on_knot(
70 client: httpx.Client,
71 knot_hostname: str,
72 repo_did: str,
73) -> dict[str, Any] | None:
74 host = knot_hostname.removeprefix("https://").rstrip("/")
75 resp = client.get(
76 f"https://{host}/xrpc/sh.tangled.repo.describeRepo",
77 params={"repoDid": repo_did},
78 timeout=20.0,
79 )
80 if resp.status_code == 404:
81 return None
82 resp.raise_for_status()
83 return resp.json()
84
85
86def normalize_tree_entries(tree: dict[str, Any]) -> list[dict[str, str]]:
87 """Flatten knot tree response into simple name/type entries."""
88 out: list[dict[str, str]] = []
89 for entry in tree.get("files") or []:
90 if not isinstance(entry, dict):
91 continue
92 name = entry.get("name")
93 if not isinstance(name, str):
94 continue
95 kind = entry.get("type")
96 if not isinstance(kind, str):
97 mode = entry.get("mode")
98 kind = "dir" if mode == "040000" else "file"
99 out.append({"name": name, "type": kind})
100 return out