feat(adapter/vcs): implement GitHubAdapter
Uses PyGithub to interact with the GitHub REST API.
- Reads GITHUB_TOKEN from env; parses owner/repo from SSH or HTTPS URL
- create_branch() creates a branch off the configured base branch
- commit() accepts dict[str, str] {path: content} or list[str] of
local paths; uses Contents API (create_file / update_file)
- create_pr() and get_pr_status() delegate to PyGithub pull-request API
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,16 +1,30 @@
|
||||
"""
|
||||
adapters/vcs/github.py
|
||||
GitHub VCS adapter — Phase 2 stub.
|
||||
GitHub VCS adapter — Phase 2 implementation.
|
||||
|
||||
TODO (Phase 2):
|
||||
- Implement create_branch() using PyGithub or gh CLI subprocess.
|
||||
- Implement commit() — stage files and push via git subprocess or API.
|
||||
- Implement create_pr() using GitHub REST API (POST /repos/{owner}/{repo}/pulls).
|
||||
- Implement get_pr_status() using GET /repos/{owner}/{repo}/pulls/{pull_number}.
|
||||
- Read repo and credentials from config/team.yaml and environment (GITHUB_TOKEN).
|
||||
Uses PyGithub (``pip install PyGithub``) to interact with the GitHub REST API.
|
||||
Reads the repository URL and base branch from the team.yaml config dict.
|
||||
|
||||
Note on commit() signature
|
||||
--------------------------
|
||||
The base class declares ``commit(files: list[str], message: str)``, which is
|
||||
insufficient for the GitHub Contents API (which requires file *content*, not
|
||||
just paths). This implementation extends the signature to accept either:
|
||||
|
||||
* ``dict[str, str]`` — ``{path: content}`` mapping (preferred; uses the API).
|
||||
* ``list[str]`` — local file paths; content is read from disk and pushed.
|
||||
|
||||
The optional ``branch`` keyword argument targets a specific branch; it
|
||||
defaults to the configured base branch.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from typing import Union
|
||||
|
||||
from github import Github, GithubException
|
||||
|
||||
from adapters.base.vcs import VCSAdapter
|
||||
|
||||
|
||||
@@ -18,34 +32,175 @@ class GitHubAdapter(VCSAdapter):
|
||||
"""
|
||||
VCS adapter for GitHub repositories.
|
||||
|
||||
Expects environment variable GITHUB_TOKEN and config values:
|
||||
run.repo — SSH or HTTPS clone URL
|
||||
run.base_branch — default base branch (e.g. "main")
|
||||
Authenticates via GITHUB_TOKEN and interacts with the GitHub REST API
|
||||
through PyGithub.
|
||||
|
||||
Environment variables
|
||||
---------------------
|
||||
GITHUB_TOKEN : Required. Personal access token or GitHub App installation token.
|
||||
|
||||
Config keys (from team.yaml)
|
||||
----------------------------
|
||||
run.repo : SSH or HTTPS clone URL (e.g. "git@github.com:org/repo.git").
|
||||
run.base_branch : Default base branch (e.g. "main").
|
||||
"""
|
||||
|
||||
def __init__(self, config: dict) -> None:
|
||||
# TODO (Phase 2): Accept loaded team.yaml config dict.
|
||||
# Extract GITHUB_TOKEN from environment.
|
||||
# Parse owner/repo from config.run.repo.
|
||||
raise NotImplementedError("GitHubAdapter.__init__ is not yet implemented.")
|
||||
"""
|
||||
Initialise the GitHub adapter.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
config : Loaded team.yaml config dict.
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If GITHUB_TOKEN is not set or the repo URL cannot be parsed.
|
||||
"""
|
||||
self._config = config
|
||||
token = os.environ.get("GITHUB_TOKEN")
|
||||
if not token:
|
||||
raise ValueError(
|
||||
"GITHUB_TOKEN environment variable is not set. "
|
||||
"Create a personal access token and export it before running the-agency."
|
||||
)
|
||||
self._g = Github(token)
|
||||
|
||||
run_cfg: dict = config.get("run", {})
|
||||
repo_url: str = run_cfg.get("repo", "")
|
||||
self._base_branch: str = run_cfg.get("base_branch", "main")
|
||||
|
||||
self._owner, self._repo_name = self._parse_repo_url(repo_url)
|
||||
self._repo = self._g.get_repo(f"{self._owner}/{self._repo_name}")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _parse_repo_url(self, url: str) -> tuple[str, str]:
|
||||
"""Parse *owner* and *repo* name from an SSH or HTTPS GitHub URL."""
|
||||
# git@github.com:owner/repo.git
|
||||
m = re.match(r"git@github\.com:([^/]+)/([^/]+?)(?:\.git)?$", url)
|
||||
if m:
|
||||
return m.group(1), m.group(2)
|
||||
# https://github.com/owner/repo[.git]
|
||||
m = re.match(r"https?://github\.com/([^/]+)/([^/]+?)(?:\.git)?/?$", url)
|
||||
if m:
|
||||
return m.group(1), m.group(2)
|
||||
raise ValueError(
|
||||
f"Cannot parse GitHub owner/repo from URL: {url!r}. "
|
||||
"Expected SSH (git@github.com:owner/repo.git) or "
|
||||
"HTTPS (https://github.com/owner/repo.git) format."
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# VCSAdapter interface
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def create_branch(self, name: str) -> None:
|
||||
# TODO (Phase 2): Create branch via GitHub API or local git subprocess.
|
||||
# Use config.run.base_branch as the branch point.
|
||||
raise NotImplementedError("GitHubAdapter.create_branch is not yet implemented.")
|
||||
"""
|
||||
Create a new branch off ``self._base_branch`` on the remote.
|
||||
|
||||
def commit(self, files: list[str], message: str) -> str:
|
||||
# TODO (Phase 2): Stage files (git add), create commit (git commit), push.
|
||||
# Return the resulting commit SHA.
|
||||
raise NotImplementedError("GitHubAdapter.commit is not yet implemented.")
|
||||
Parameters
|
||||
----------
|
||||
name : New branch name (e.g. "feat/webhook-ingestion").
|
||||
"""
|
||||
base_ref = self._repo.get_git_ref(f"heads/{self._base_branch}")
|
||||
self._repo.create_git_ref(f"refs/heads/{name}", base_ref.object.sha)
|
||||
|
||||
def commit(
|
||||
self,
|
||||
files: Union[dict[str, str], list[str]],
|
||||
message: str,
|
||||
branch: str | None = None,
|
||||
) -> str:
|
||||
"""
|
||||
Commit files to the repository via the GitHub Contents API.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
files : Either a ``dict[path, content]`` mapping (preferred), or a
|
||||
``list[path]`` of local file paths whose content is read from
|
||||
disk.
|
||||
message : Commit message.
|
||||
branch : Target branch. Defaults to ``self._base_branch``.
|
||||
|
||||
Returns
|
||||
-------
|
||||
SHA of the last created/updated commit, or empty string if no files
|
||||
were committed.
|
||||
"""
|
||||
target_branch = branch or self._base_branch
|
||||
|
||||
# Normalise to {path: content}
|
||||
if isinstance(files, list):
|
||||
files_dict: dict[str, str] = {}
|
||||
for path in files:
|
||||
with open(path, "r", encoding="utf-8") as fh:
|
||||
files_dict[path] = fh.read()
|
||||
else:
|
||||
files_dict = files
|
||||
|
||||
last_sha: str = ""
|
||||
for path, content in files_dict.items():
|
||||
try:
|
||||
existing = self._repo.get_contents(path, ref=target_branch)
|
||||
result = self._repo.update_file(
|
||||
path=path,
|
||||
message=message,
|
||||
content=content,
|
||||
sha=existing.sha, # type: ignore[union-attr]
|
||||
branch=target_branch,
|
||||
)
|
||||
except GithubException:
|
||||
# File does not exist yet — create it
|
||||
result = self._repo.create_file(
|
||||
path=path,
|
||||
message=message,
|
||||
content=content,
|
||||
branch=target_branch,
|
||||
)
|
||||
last_sha = result["commit"].sha
|
||||
|
||||
return last_sha
|
||||
|
||||
def create_pr(self, title: str, body: str, head: str, base: str) -> str:
|
||||
# TODO (Phase 2): POST to GitHub API /repos/{owner}/{repo}/pulls.
|
||||
# Return the HTML URL of the created PR.
|
||||
raise NotImplementedError("GitHubAdapter.create_pr is not yet implemented.")
|
||||
"""
|
||||
Open a pull request on GitHub.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
title : PR title.
|
||||
body : PR description / body markdown.
|
||||
head : Head branch name (the branch with changes).
|
||||
base : Base branch name (e.g. "main").
|
||||
|
||||
Returns
|
||||
-------
|
||||
HTML URL of the created pull request.
|
||||
"""
|
||||
pr = self._repo.create_pull(
|
||||
title=title,
|
||||
body=body,
|
||||
head=head,
|
||||
base=base,
|
||||
)
|
||||
return pr.html_url
|
||||
|
||||
def get_pr_status(self, pr_id: str) -> str:
|
||||
# TODO (Phase 2): GET /repos/{owner}/{repo}/pulls/{number}.
|
||||
# Map GitHub PR state ("open", "closed") + merged flag to
|
||||
# our schema: "open" | "merged" | "closed".
|
||||
raise NotImplementedError("GitHubAdapter.get_pr_status is not yet implemented.")
|
||||
"""
|
||||
Fetch the current status of a pull request.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
pr_id : Pull request number as a string (e.g. "42").
|
||||
|
||||
Returns
|
||||
-------
|
||||
One of: "open" | "merged" | "closed".
|
||||
"""
|
||||
pr = self._repo.get_pull(int(pr_id))
|
||||
if pr.merged:
|
||||
return "merged"
|
||||
return pr.state # "open" or "closed"
|
||||
|
||||
Reference in New Issue
Block a user