radicle.py
1 """Radicle integration service - wraps rad CLI for direct access.""" 2 3 import asyncio 4 import json 5 import subprocess 6 from pathlib import Path 7 from typing import Optional 8 from dataclasses import dataclass 9 10 from config import get_settings 11 12 settings = get_settings() 13 14 15 @dataclass 16 class RadicleRepo: 17 """Radicle repository metadata.""" 18 19 rid: str # Radicle ID (rad:...) 20 name: str 21 description: str 22 default_branch: str 23 delegates: list[str] 24 25 26 @dataclass 27 class RadiclePatch: 28 """Radicle patch (PR equivalent).""" 29 30 patch_id: str 31 title: str 32 author: str 33 state: str # open, merged, closed 34 created_at: str 35 updated_at: str 36 target_branch: str 37 revisions: int 38 39 40 @dataclass 41 class FileEntry: 42 """File or directory entry in repo tree.""" 43 44 name: str 45 path: str 46 type: str # "file" or "dir" 47 size: Optional[int] = None 48 49 50 class RadicleService: 51 """Service for interacting with Radicle via rad CLI.""" 52 53 def __init__(self): 54 self.rad_path = settings.rad_path 55 self.radicle_home = Path(settings.radicle_home).expanduser() 56 57 async def _run_rad(self, *args: str, cwd: Optional[Path] = None) -> tuple[int, str, str]: 58 """Run rad command and return (returncode, stdout, stderr).""" 59 cmd = [self.rad_path] + list(args) 60 61 process = await asyncio.create_subprocess_exec( 62 *cmd, 63 stdout=asyncio.subprocess.PIPE, 64 stderr=asyncio.subprocess.PIPE, 65 cwd=cwd, 66 ) 67 68 stdout, stderr = await process.communicate() 69 return process.returncode, stdout.decode(), stderr.decode() 70 71 async def list_repos(self) -> list[RadicleRepo]: 72 """List all tracked Radicle repositories.""" 73 returncode, stdout, stderr = await self._run_rad("ls", "--json") 74 75 if returncode != 0: 76 raise RuntimeError(f"rad ls failed: {stderr}") 77 78 repos = [] 79 for line in stdout.strip().split("\n"): 80 if not line: 81 continue 82 try: 83 data = json.loads(line) 84 repos.append( 85 RadicleRepo( 86 rid=data.get("rid", ""), 87 name=data.get("name", ""), 88 description=data.get("description", ""), 89 default_branch=data.get("defaultBranch", "main"), 90 delegates=data.get("delegates", []), 91 ) 92 ) 93 except json.JSONDecodeError: 94 continue 95 96 return repos 97 98 async def get_repo(self, rid: str) -> Optional[RadicleRepo]: 99 """Get repository by Radicle ID.""" 100 returncode, stdout, stderr = await self._run_rad("inspect", rid, "--json") 101 102 if returncode != 0: 103 return None 104 105 try: 106 data = json.loads(stdout) 107 return RadicleRepo( 108 rid=data.get("rid", rid), 109 name=data.get("name", ""), 110 description=data.get("description", ""), 111 default_branch=data.get("defaultBranch", "main"), 112 delegates=data.get("delegates", []), 113 ) 114 except json.JSONDecodeError: 115 return None 116 117 async def get_tree( 118 self, rid: str, ref: str = "HEAD", path: str = "" 119 ) -> list[FileEntry]: 120 """Get file tree for a repository at given ref and path.""" 121 # Use git ls-tree through rad 122 returncode, stdout, stderr = await self._run_rad( 123 "inspect", rid, "--tree", ref, "--path", path 124 ) 125 126 if returncode != 0: 127 # Fallback: try direct git access 128 repo_path = self._get_repo_path(rid) 129 if repo_path: 130 return await self._git_ls_tree(repo_path, ref, path) 131 raise RuntimeError(f"Failed to get tree: {stderr}") 132 133 entries = [] 134 for line in stdout.strip().split("\n"): 135 if not line: 136 continue 137 parts = line.split("\t") 138 if len(parts) >= 2: 139 mode_type = parts[0].split() 140 name = parts[1] 141 entry_type = "dir" if mode_type[1] == "tree" else "file" 142 entries.append( 143 FileEntry( 144 name=name, 145 path=f"{path}/{name}".lstrip("/"), 146 type=entry_type, 147 ) 148 ) 149 150 return entries 151 152 async def _git_ls_tree( 153 self, repo_path: Path, ref: str, path: str 154 ) -> list[FileEntry]: 155 """Direct git ls-tree access for repository.""" 156 cmd = ["git", "ls-tree", ref] 157 if path: 158 cmd.append(path) 159 160 process = await asyncio.create_subprocess_exec( 161 *cmd, 162 stdout=asyncio.subprocess.PIPE, 163 stderr=asyncio.subprocess.PIPE, 164 cwd=repo_path, 165 ) 166 167 stdout, stderr = await process.communicate() 168 if process.returncode != 0: 169 raise RuntimeError(f"git ls-tree failed: {stderr.decode()}") 170 171 entries = [] 172 for line in stdout.decode().strip().split("\n"): 173 if not line: 174 continue 175 # Format: <mode> <type> <object> <file> 176 parts = line.split("\t") 177 if len(parts) >= 2: 178 mode_type_obj = parts[0].split() 179 name = parts[1] 180 entry_type = "dir" if mode_type_obj[1] == "tree" else "file" 181 entries.append( 182 FileEntry( 183 name=name, 184 path=f"{path}/{name}".lstrip("/"), 185 type=entry_type, 186 ) 187 ) 188 189 return entries 190 191 async def get_blob(self, rid: str, ref: str, path: str) -> Optional[str]: 192 """Get file content from repository.""" 193 repo_path = self._get_repo_path(rid) 194 if not repo_path: 195 return None 196 197 cmd = ["git", "show", f"{ref}:{path}"] 198 process = await asyncio.create_subprocess_exec( 199 *cmd, 200 stdout=asyncio.subprocess.PIPE, 201 stderr=asyncio.subprocess.PIPE, 202 cwd=repo_path, 203 ) 204 205 stdout, stderr = await process.communicate() 206 if process.returncode != 0: 207 return None 208 209 return stdout.decode() 210 211 async def list_patches(self, rid: str) -> list[RadiclePatch]: 212 """List patches (PRs) for a repository.""" 213 returncode, stdout, stderr = await self._run_rad( 214 "patch", "list", "--repo", rid, "--json" 215 ) 216 217 if returncode != 0: 218 return [] 219 220 patches = [] 221 for line in stdout.strip().split("\n"): 222 if not line: 223 continue 224 try: 225 data = json.loads(line) 226 patches.append( 227 RadiclePatch( 228 patch_id=data.get("id", ""), 229 title=data.get("title", ""), 230 author=data.get("author", {}).get("id", ""), 231 state=data.get("state", "open"), 232 created_at=data.get("createdAt", ""), 233 updated_at=data.get("updatedAt", ""), 234 target_branch=data.get("target", "main"), 235 revisions=len(data.get("revisions", [])), 236 ) 237 ) 238 except json.JSONDecodeError: 239 continue 240 241 return patches 242 243 async def get_patch(self, rid: str, patch_id: str) -> Optional[RadiclePatch]: 244 """Get patch details.""" 245 returncode, stdout, stderr = await self._run_rad( 246 "patch", "show", patch_id, "--repo", rid, "--json" 247 ) 248 249 if returncode != 0: 250 return None 251 252 try: 253 data = json.loads(stdout) 254 return RadiclePatch( 255 patch_id=data.get("id", patch_id), 256 title=data.get("title", ""), 257 author=data.get("author", {}).get("id", ""), 258 state=data.get("state", "open"), 259 created_at=data.get("createdAt", ""), 260 updated_at=data.get("updatedAt", ""), 261 target_branch=data.get("target", "main"), 262 revisions=len(data.get("revisions", [])), 263 ) 264 except json.JSONDecodeError: 265 return None 266 267 async def get_patch_diff(self, rid: str, patch_id: str) -> Optional[str]: 268 """Get diff for a patch.""" 269 returncode, stdout, stderr = await self._run_rad( 270 "patch", "diff", patch_id, "--repo", rid 271 ) 272 273 if returncode != 0: 274 return None 275 276 return stdout 277 278 def _get_repo_path(self, rid: str) -> Optional[Path]: 279 """Get local path for a Radicle repository.""" 280 # Radicle stores repos in ~/.radicle/storage/<rid>/ 281 # The rid format is rad:z... - we need to extract the ID part 282 if rid.startswith("rad:"): 283 repo_id = rid[4:] 284 else: 285 repo_id = rid 286 287 storage_path = self.radicle_home / "storage" / repo_id 288 if storage_path.exists(): 289 return storage_path 290 291 return None 292 293 294 # Singleton instance 295 _radicle_service: Optional[RadicleService] = None 296 297 298 def get_radicle_service() -> RadicleService: 299 """Get Radicle service singleton.""" 300 global _radicle_service 301 if _radicle_service is None: 302 _radicle_service = RadicleService() 303 return _radicle_service