diff --git a/agent/utils/github.py b/agent/utils/git_utils.py similarity index 50% rename from agent/utils/github.py rename to agent/utils/git_utils.py index 2932674..1764b08 100644 --- a/agent/utils/github.py +++ b/agent/utils/git_utils.py @@ -1,19 +1,11 @@ -"""GitHub API and git utilities.""" +"""Git utilities for repository operations.""" from __future__ import annotations -import logging import shlex -import httpx from deepagents.backends.protocol import ExecuteResponse, SandboxBackendProtocol -logger = logging.getLogger(__name__) - -# HTTP status codes -HTTP_CREATED = 201 -HTTP_UNPROCESSABLE_ENTITY = 422 - def _run_git( sandbox_backend: SandboxBackendProtocol, repo_dir: str, command: str @@ -156,164 +148,3 @@ def git_push( return _git_with_credentials(sandbox_backend, repo_dir, f"push origin {safe_branch}") finally: cleanup_git_credentials(sandbox_backend) - - -async def create_github_pr( - repo_owner: str, - repo_name: str, - github_token: str, - title: str, - head_branch: str, - base_branch: str, - body: str, -) -> tuple[str | None, int | None, bool]: - """Create a draft GitHub pull request via the API. - - Args: - repo_owner: Repository owner (e.g., "langchain-ai") - repo_name: Repository name (e.g., "deepagents") - github_token: GitHub access token - title: PR title - head_branch: Source branch name - base_branch: Target branch name - body: PR description - - Returns: - Tuple of (pr_url, pr_number, pr_existing) if successful, (None, None, False) otherwise - """ - pr_payload = { - "title": title, - "head": head_branch, - "base": base_branch, - "body": body, - "draft": True, - } - - logger.info( - "Creating PR: head=%s, base=%s, repo=%s/%s", - head_branch, - base_branch, - repo_owner, - repo_name, - ) - - async with httpx.AsyncClient() as http_client: - try: - pr_response = await http_client.post( - f"https://api.github.com/repos/{repo_owner}/{repo_name}/pulls", - headers={ - "Authorization": f"Bearer {github_token}", - "Accept": "application/vnd.github+json", - "X-GitHub-Api-Version": "2022-11-28", - }, - json=pr_payload, - ) - - pr_data = pr_response.json() - - if pr_response.status_code == HTTP_CREATED: - pr_url = pr_data.get("html_url") - pr_number = pr_data.get("number") - logger.info("PR created successfully: %s", pr_url) - return pr_url, pr_number, False - - if pr_response.status_code == HTTP_UNPROCESSABLE_ENTITY: - logger.error("GitHub API validation error (422): %s", pr_data.get("message")) - existing = await _find_existing_pr( - http_client=http_client, - repo_owner=repo_owner, - repo_name=repo_name, - github_token=github_token, - head_branch=head_branch, - ) - if existing: - logger.info("Using existing PR for head branch: %s", existing[0]) - return existing[0], existing[1], True - else: - logger.error( - "GitHub API error (%s): %s", - pr_response.status_code, - pr_data.get("message"), - ) - - if "errors" in pr_data: - logger.error("GitHub API errors detail: %s", pr_data.get("errors")) - - return None, None, False - - except httpx.HTTPError: - logger.exception("Failed to create PR via GitHub API") - return None, None, False - - -async def _find_existing_pr( - http_client: httpx.AsyncClient, - repo_owner: str, - repo_name: str, - github_token: str, - head_branch: str, -) -> tuple[str | None, int | None]: - """Find an existing PR for the given head branch.""" - headers = { - "Authorization": f"Bearer {github_token}", - "Accept": "application/vnd.github+json", - "X-GitHub-Api-Version": "2022-11-28", - } - head_ref = f"{repo_owner}:{head_branch}" - for state in ("open", "all"): - response = await http_client.get( - f"https://api.github.com/repos/{repo_owner}/{repo_name}/pulls", - headers=headers, - params={"head": head_ref, "state": state, "per_page": 1}, - ) - if response.status_code != 200: # noqa: PLR2004 - continue - data = response.json() - if not data: - continue - pr = data[0] - return pr.get("html_url"), pr.get("number") - return None, None - - -async def get_github_default_branch( - repo_owner: str, - repo_name: str, - github_token: str, -) -> str: - """Get the default branch of a GitHub repository via the API. - - Args: - repo_owner: Repository owner (e.g., "langchain-ai") - repo_name: Repository name (e.g., "deepagents") - github_token: GitHub access token - - Returns: - The default branch name (e.g., "main" or "master") - """ - try: - async with httpx.AsyncClient() as http_client: - response = await http_client.get( - f"https://api.github.com/repos/{repo_owner}/{repo_name}", - headers={ - "Authorization": f"Bearer {github_token}", - "Accept": "application/vnd.github+json", - "X-GitHub-Api-Version": "2022-11-28", - }, - ) - - if response.status_code == 200: # noqa: PLR2004 - repo_data = response.json() - default_branch = repo_data.get("default_branch", "main") - logger.debug("Got default branch from GitHub API: %s", default_branch) - return default_branch - - logger.warning( - "Failed to get repo info from GitHub API (%s), falling back to 'main'", - response.status_code, - ) - return "main" - - except httpx.HTTPError: - logger.exception("Failed to get default branch from GitHub API, falling back to 'main'") - return "main"