refactor: extract git utilities from github.py into git_utils.py
This commit is contained in:
머니페니
parent
33db8eb7b0
commit
0e5672f648
@ -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"
|
||||
Loading…
x
Reference in New Issue
Block a user