Spaces:
Sleeping
Sleeping
| """ | |
| Thin GitHub REST client for token validation + publishing a repo. | |
| ponytail: kept tiny and HTTP-call-shaped (module-level httpx.get/post/put) so tests can | |
| patch `integrations.github_client.httpx.*` without any network. Never logs or returns the PAT. | |
| """ | |
| import base64 | |
| import logging | |
| import httpx | |
| logger = logging.getLogger(__name__) | |
| GITHUB_API = "https://api.github.com" | |
| _TIMEOUT = 15.0 | |
| class GitHubError(RuntimeError): | |
| """A GitHub API call failed (network or non-2xx). Message is safe to surface — no token.""" | |
| def _headers(token: str) -> dict[str, str]: | |
| return { | |
| "Authorization": f"token {token}", | |
| "Accept": "application/vnd.github+json", | |
| "X-GitHub-Api-Version": "2022-11-28", | |
| } | |
| def validate_token(token: str) -> str: | |
| """Return the GitHub login for a PAT. Raise ValueError if the token is invalid/unauthorized.""" | |
| try: | |
| resp = httpx.get(f"{GITHUB_API}/user", headers=_headers(token), timeout=_TIMEOUT) | |
| except httpx.HTTPError as exc: # network failure — not the user's fault, but token unverifiable | |
| raise ValueError("Could not reach GitHub to validate the token") from exc | |
| if resp.status_code != 200: | |
| raise ValueError("Invalid GitHub token") | |
| login = (resp.json() or {}).get("login") | |
| if not login: | |
| raise ValueError("Invalid GitHub token") | |
| return login | |
| def _ensure_repo(token: str, login: str, name: str) -> dict: | |
| """Get the repo if it exists, else create it (public). Returns the repo JSON.""" | |
| h = _headers(token) | |
| try: | |
| existing = httpx.get(f"{GITHUB_API}/repos/{login}/{name}", headers=h, timeout=_TIMEOUT) | |
| if existing.status_code == 200: | |
| return existing.json() | |
| created = httpx.post( | |
| f"{GITHUB_API}/user/repos", | |
| headers=h, | |
| json={"name": name, "private": False, "auto_init": True, | |
| "description": "Generated by AppSmith / CodyBuddy Studio"}, | |
| timeout=_TIMEOUT, | |
| ) | |
| except httpx.HTTPError as exc: | |
| raise GitHubError("GitHub request failed while creating the repository") from exc | |
| if created.status_code not in (200, 201): | |
| raise GitHubError(f"GitHub could not create the repository (status {created.status_code})") | |
| return created.json() | |
| def _put_file(token: str, owner: str, repo: str, path: str, content: str, branch: str) -> None: | |
| """Create or update a single file via the contents API (handles existing-file sha).""" | |
| h = _headers(token) | |
| url = f"{GITHUB_API}/repos/{owner}/{repo}/contents/{path}" | |
| try: | |
| sha = None | |
| head = httpx.get(url, headers=h, params={"ref": branch}, timeout=_TIMEOUT) | |
| if head.status_code == 200: | |
| sha = (head.json() or {}).get("sha") | |
| payload: dict = { | |
| "message": f"chore: add {path}", | |
| "content": base64.b64encode(content.encode()).decode(), | |
| "branch": branch, | |
| } | |
| if sha: | |
| payload["sha"] = sha | |
| resp = httpx.put(url, headers=h, json=payload, timeout=_TIMEOUT) | |
| except httpx.HTTPError as exc: | |
| raise GitHubError(f"GitHub request failed while writing {path}") from exc | |
| if resp.status_code not in (200, 201): | |
| raise GitHubError(f"GitHub rejected writing {path} (status {resp.status_code})") | |
| def publish_files(token: str, repo_name: str, files: dict[str, str]) -> str: | |
| """Create-or-update `repo_name` under the authed user and push all files. Returns the repo HTML URL. | |
| Raises ValueError if the token is invalid, GitHubError on any API failure. | |
| """ | |
| login = validate_token(token) # also resolves the owner login | |
| repo = _ensure_repo(token, login, repo_name) | |
| branch = repo.get("default_branch") or "main" | |
| repo_url = repo.get("html_url") or f"https://github.com/{login}/{repo_name}" | |
| for path, content in files.items(): | |
| _put_file(token, login, repo_name, path.lstrip("/"), content, branch) | |
| return repo_url | |