From 6bc3da693767b6c39b1f6988294fea5cfd083a96 Mon Sep 17 00:00:00 2001 From: Your Name Date: Tue, 13 May 2025 14:53:56 -0400 Subject: [PATCH] Refactor GitHubClient: replace PyGithub with requests for API interactions, enhance error handling, and streamline repository creation from templates --- template_automation/github_client.py | 590 +++++++++++++-------------- 1 file changed, 289 insertions(+), 301 deletions(-) diff --git a/template_automation/github_client.py b/template_automation/github_client.py index 242a060..9fa5057 100644 --- a/template_automation/github_client.py +++ b/template_automation/github_client.py @@ -1,21 +1,17 @@ """GitHub client module for template automation. This module provides the GitHubClient class which handles all interactions with the GitHub API -for template repository automation. +for template repository automation using the requests library directly. """ import base64 +import json import logging import time +import urllib.parse from typing import List, Optional, Dict, Any, Union -from github import Github, GithubException -from github.Repository import Repository -from github.ContentFile import ContentFile -from github.Organization import Organization -from github.Team import Team -from github.PullRequest import PullRequest -from github.Workflow import Workflow +import requests logger = logging.getLogger(__name__) @@ -34,8 +30,6 @@ class GitHubClient: org_name (str): GitHub organization name commit_author_name (str): Name to use for automated commits commit_author_email (str): Email to use for automated commits - client (Github): PyGithub client instance - org (Organization): GitHub organization instance verify_ssl (bool): Whether to verify SSL certificates Example: @@ -74,211 +68,66 @@ def __init__( commit_author_name: Name to use for automated commits commit_author_email: Email to use for automated commits verify_ssl: Whether to verify SSL certificates - - Raises: - GithubException: If authentication fails or org doesn't exist """ - self.api_base_url = api_base_url + self.api_base_url = api_base_url.rstrip('/') self.token = token self.org_name = org_name self.commit_author_name = commit_author_name self.commit_author_email = commit_author_email self.verify_ssl = verify_ssl - # Set environment variable for PyGithub to allow any hostname (for enterprise GitHub) - # This is needed before initializing the client - import os - os.environ["GITHUB_ALLOW_HOSTNAME"] = "TRUE" - - try: - # Try to use modern auth approach if available - from github import Auth - auth = Auth.Token(token) - self.client = Github( - auth=auth, - base_url=api_base_url, - verify=verify_ssl, - per_page=100, # Optimize API call efficiency - timeout=30, # Set a reasonable timeout - retry=10 # Enable retries for transient issues - ) - except ImportError: - # Fall back to older authentication method if Auth module isn't available - self.client = Github( - login_or_token=token, - base_url=api_base_url, - verify=verify_ssl, - per_page=100, # Optimize API call efficiency - timeout=30, # Set a reasonable timeout - ) + # Create session for connection reuse + self.session = requests.Session() + self.session.headers.update({ + 'Authorization': f'token {token}', + 'Accept': 'application/vnd.github.v3+json', + 'User-Agent': 'Template-Automation-Lambda' + }) - try: - self.org = self.client.get_organization(org_name) - logger.info(f"Initialized GitHub client for org: {org_name} (SSL verify: {verify_ssl})") - except GithubException as e: - logger.error(f"Failed to initialize GitHub client: {str(e)}") - raise + # Log initialization + logger.info(f"Initialized GitHub client for org: {org_name} (SSL verify: {verify_ssl})") - def create_repository_from_template( - self, - template_repo_name: str, - new_repo_name: str, - private: bool = True, - description: Optional[str] = None, - topics: Optional[List[str]] = None - ) -> Repository: - """Create a new repository from a template. + def _request(self, method: str, url: str, **kwargs) -> Dict[str, Any]: + """Make a request to the GitHub API. Args: - template_repo_name: Name of the template repository - new_repo_name: Name for the new repository - private: Whether the new repository should be private - description: Description for the new repository - topics: List of topics to add to the repository + method: HTTP method (GET, POST, PATCH, PUT, DELETE) + url: URL path or full URL to request + **kwargs: Additional arguments to pass to requests Returns: - The newly created repository + Response data as a dictionary Raises: - GithubException: If template doesn't exist or repo creation fails + requests.exceptions.RequestException: On request errors """ - template_repo = self.org.get_repo(template_repo_name) + # Prepend base URL if not already an absolute URL + if not url.startswith('http'): + url = f"{self.api_base_url}{url}" - # Create repository from template - new_repo = self.org.create_repository_from_template( - name=new_repo_name, - template_repository=template_repo, - private=private, - description=description or f"Repository created from template: {template_repo_name}" - ) + # Set SSL verification + kwargs['verify'] = self.verify_ssl - # Add topics if provided - if topics: - new_repo.replace_topics(topics) - - logger.info(f"Created new repository: {new_repo_name} from template: {template_repo_name}") - return new_repo - - def set_team_access(self, repo: Repository, team_slug: str, permission: str = "admin") -> None: - """Give a team access to a repository. + # Log the request + logger.debug(f"GitHub API {method} request: {url}") - Args: - repo: The repository to grant access to - team_slug: The team's slug identifier - permission: The permission level to grant (pull, push, admin) - - Raises: - GithubException: If team doesn't exist or permission grant fails - """ - try: - team = self.org.get_team_by_slug(team_slug) - team.add_to_repos(repo) - team.set_repo_permission(repo, permission) - logger.info(f"Granted {permission} access to team {team_slug} for repo {repo.name}") - except GithubException as e: - logger.error(f"Failed to set team access: {e}") - raise - - def write_file( - self, - repo: Repository, - path: str, - content: str, - branch: str = "main", - commit_message: Optional[str] = None - ) -> ContentFile: - """Write or update a file in a repository. + # Make the request + response = self.session.request(method, url, **kwargs) - Args: - repo: The repository to write to - path: Path where to create/update the file - content: Content to write to the file - branch: Branch to commit to - commit_message: Commit message to use - - Returns: - The created/updated file content - - Raises: - GithubException: If file operation fails - """ - try: - # Convert content to base64 - content_bytes = content.encode("utf-8") - content_base64 = base64.b64encode(content_bytes).decode("utf-8") - - # Try to get existing file - try: - file = repo.get_contents(path, ref=branch) - # Update existing file - result = repo.update_file( - path=path, - message=commit_message or f"Update {path}", - content=content_base64, - sha=file.sha, - branch=branch, - committer={ - "name": self.commit_author_name, - "email": self.commit_author_email - } - ) - logger.info(f"Updated file {path} in repo {repo.name}") - return result["content"] - except GithubException as e: - if e.status != 404: # Only handle "not found" errors - raise - - # Create new file - result = repo.create_file( - path=path, - message=commit_message or f"Create {path}", - content=content_base64, - branch=branch, - committer={ - "name": self.commit_author_name, - "email": self.commit_author_email - } - ) - logger.info(f"Created new file {path} in repo {repo.name}") - return result["content"] - - except GithubException as e: - logger.error(f"Failed to write file {path}: {e}") - raise - - def read_file( - self, - repo: Repository, - path: str, - ref: str = "main" - ) -> str: - """Read a file from a repository. + # Raise exception for error status codes + response.raise_for_status() - Args: - repo: The repository to read from - path: Path to the file to read - ref: Git reference (branch, tag, commit) to read from - - Returns: - The file contents as a string - - Raises: - GithubException: If file doesn't exist or read fails - """ - try: - file = repo.get_contents(path, ref=ref) - content = base64.b64decode(file.content).decode("utf-8") - return content - except GithubException as e: - logger.error(f"Failed to read file {path}: {e}") - raise + # Return JSON data for non-empty responses + if response.text: + return response.json() + return {} def get_repository( self, repo_name: str, create: bool = False, owning_team: Optional[str] = None - ) -> Repository: + ) -> Dict[str, Any]: """Get or create a GitHub repository with optional team permissions. Args: @@ -287,52 +136,65 @@ def get_repository( owning_team: The name of the GitHub team to grant admin access Returns: - The repository object - - Raises: - GithubException: If repository operations fail + The repository data """ try: - try: - repo = self.org.get_repo(repo_name) - logger.info(f"Found existing repository: {repo_name}") + # Try to get the repository + url = f"/repos/{self.org_name}/{repo_name}" + repo = self._request("GET", url) + logger.info(f"Found existing repository: {repo_name}") + + if owning_team: + self.set_team_permission(repo_name, owning_team, "admin") + + return repo + except requests.exceptions.HTTPError as e: + if e.response.status_code == 404 and create: + logger.info(f"Creating repository {repo_name}") + + # Create a new repository + url = f"/orgs/{self.org_name}/repos" + repo = self._request("POST", url, json={ + "name": repo_name, + "private": True, + "auto_init": True, + "allow_squash_merge": True, + "allow_merge_commit": True, + "allow_rebase_merge": True, + "delete_branch_on_merge": True + }) + + # Wait for repository initialization + max_retries = 100 + retry_delay = 1 + for _ in range(max_retries): + try: + self.get_branch(repo_name, "main") + break + except requests.exceptions.HTTPError: + time.sleep(retry_delay) + else: + raise Exception(f"Repository {repo_name} initialization timed out") + if owning_team: self.set_team_permission(repo_name, owning_team, "admin") - return repo - except GithubException as e: - if e.status == 404 and create: - logger.info(f"Creating repository {repo_name}") - repo = self.org.create_repo( - name=repo_name, - private=True, - auto_init=True, - allow_squash_merge=True, - allow_merge_commit=True, - allow_rebase_merge=True, - delete_branch_on_merge=True - ) - # Wait for repository initialization - max_retries = 100 - retry_delay = 1 - for _ in range(max_retries): - try: - repo.get_branch("main") - break - except GithubException: - time.sleep(retry_delay) - else: - raise Exception(f"Repository {repo_name} initialization timed out") - - if owning_team: - self.set_team_permission(repo_name, owning_team, "admin") - return repo - raise - except GithubException as e: - error_message = f"GitHub API error: {str(e)}" - logger.error(error_message) + return repo raise + def get_branch(self, repo_name: str, branch_name: str) -> Dict[str, Any]: + """Get branch information. + + Args: + repo_name: Name of the repository + branch_name: Name of the branch + + Returns: + Branch data + """ + url = f"/repos/{self.org_name}/{repo_name}/branches/{branch_name}" + return self._request("GET", url) + def get_default_branch(self, repo_name: str) -> str: """Get the default branch name of a repository. @@ -342,8 +204,8 @@ def get_default_branch(self, repo_name: str) -> str: Returns: Default branch name (usually 'main' or 'master') """ - repo = self.org.get_repo(repo_name) - return repo.default_branch + repo = self.get_repository(repo_name) + return repo["default_branch"] def create_branch(self, repo_name: str, branch_name: str, from_ref: str = "main") -> None: """Create a new branch in the repository. @@ -352,24 +214,142 @@ def create_branch(self, repo_name: str, branch_name: str, from_ref: str = "main" repo_name: Name of the repository branch_name: Name of the branch to create from_ref: Reference to create branch from + """ + # Get the SHA of the source branch + source_branch = self.get_branch(repo_name, from_ref) + commit_sha = source_branch["commit"]["sha"] + + # Create the new branch + url = f"/repos/{self.org_name}/{repo_name}/git/refs" + self._request("POST", url, json={ + "ref": f"refs/heads/{branch_name}", + "sha": commit_sha + }) + + logger.info(f"Created branch {branch_name} in {repo_name}") + + def create_reference(self, repo_name: str, ref: str, sha: str) -> None: + """Create a Git reference. + + Args: + repo_name: Name of the repository + ref: The name of the reference + sha: The SHA1 value to set this reference to + """ + url = f"/repos/{self.org_name}/{repo_name}/git/refs" + self._request("POST", url, json={ + "ref": ref, + "sha": sha + }) + + logger.info(f"Created reference {ref} in {repo_name}") + + def update_reference(self, repo_name: str, ref: str, sha: str, force: bool = False) -> None: + """Update a Git reference. + + Args: + repo_name: Name of the repository + ref: The name of the reference without 'refs/' prefix + sha: The SHA1 value to set this reference to + force: Force update if not a fast-forward update + """ + url = f"/repos/{self.org_name}/{repo_name}/git/refs/{ref}" + self._request("PATCH", url, json={ + "sha": sha, + "force": force + }) + + logger.info(f"Updated reference {ref} in {repo_name}") + + def write_file( + self, + repo: Dict[str, Any], + path: str, + content: str, + branch: str = "main", + commit_message: Optional[str] = None + ) -> Dict[str, Any]: + """Write or update a file in a repository. + + Args: + repo: The repository object + path: Path where to create/update the file + content: Content to write to the file + branch: Branch to commit to + commit_message: Commit message to use - Raises: - GithubException: If branch creation fails + Returns: + The created/updated file content """ - repo = self.org.get_repo(repo_name) - source = repo.get_branch(from_ref) + repo_name = repo["name"] + content_bytes = content.encode("utf-8") + content_base64 = base64.b64encode(content_bytes).decode("utf-8") + # Try to get the existing file to check if it exists try: - repo.create_git_ref( - ref=f"refs/heads/{branch_name}", - sha=source.commit.sha - ) - logger.info(f"Created branch {branch_name} in {repo_name}") - except GithubException as e: - error_message = f"Failed to create branch {branch_name}: {str(e)}" - logger.error(error_message) + file = self.get_file_contents(repo_name, path, branch) + # Update existing file + url = f"/repos/{self.org_name}/{repo_name}/contents/{path}" + result = self._request("PUT", url, json={ + "message": commit_message or f"Update {path}", + "content": content_base64, + "sha": file["sha"], + "branch": branch, + "committer": { + "name": self.commit_author_name, + "email": self.commit_author_email + } + }) + logger.info(f"Updated file {path} in repo {repo_name}") + return result["content"] + except requests.exceptions.HTTPError as e: + if e.response.status_code == 404: + # Create new file + url = f"/repos/{self.org_name}/{repo_name}/contents/{path}" + result = self._request("PUT", url, json={ + "message": commit_message or f"Create {path}", + "content": content_base64, + "branch": branch, + "committer": { + "name": self.commit_author_name, + "email": self.commit_author_email + } + }) + logger.info(f"Created new file {path} in repo {repo_name}") + return result["content"] raise + def get_file_contents(self, repo_name: str, path: str, ref: str = "main") -> Dict[str, Any]: + """Get the contents of a file in a repository. + + Args: + repo_name: Name of the repository + path: Path to the file + ref: Branch, tag, or commit SHA + + Returns: + File data + """ + url = f"/repos/{self.org_name}/{repo_name}/contents/{path}" + params = {"ref": ref} + return self._request("GET", url, params=params) + + def read_file(self, repo: Dict[str, Any], path: str, ref: str = "main") -> str: + """Read a file from a repository. + + Args: + repo: The repository object + path: Path to the file to read + ref: Git reference (branch, tag, commit) to read from + + Returns: + The file contents as a string + """ + repo_name = repo["name"] + file = self.get_file_contents(repo_name, path, ref) + content = base64.b64decode(file["content"]).decode("utf-8") + return content + def create_pull_request( self, repo_name: str, @@ -377,7 +357,7 @@ def create_pull_request( body: str, head_branch: str, base_branch: str = "main" - ) -> Any: + ) -> Dict[str, Any]: """Create a pull request in a repository. Args: @@ -389,25 +369,18 @@ def create_pull_request( Returns: The created pull request object - - Raises: - GithubException: If pull request creation fails """ - try: - repo = self.org.get_repo(repo_name) - pr = repo.create_pull( - title=title, - body=body, - head=head_branch, - base=base_branch, - maintainer_can_modify=True - ) - logger.info(f"Created PR #{pr.number} in {repo_name}: {title}") - return pr - except GithubException as e: - error_message = f"Failed to create pull request: {str(e)}" - logger.error(error_message) - raise + url = f"/repos/{self.org_name}/{repo_name}/pulls" + pr = self._request("POST", url, json={ + "title": title, + "body": body, + "head": head_branch, + "base": base_branch, + "maintainer_can_modify": True + }) + + logger.info(f"Created PR #{pr['number']} in {repo_name}: {title}") + return pr def trigger_workflow( self, @@ -423,26 +396,16 @@ def trigger_workflow( workflow_id: ID or filename of the workflow ref: Git reference to run the workflow on inputs: Input parameters for the workflow - - Raises: - GithubException: If workflow dispatch fails """ - try: - repo = self.org.get_repo(repo_name) - workflow = repo.get_workflow(workflow_id) - - # Convert inputs to GitHub's expected format - workflow_inputs = inputs if inputs is not None else {} - - workflow.create_dispatch( - ref=ref, - inputs=workflow_inputs - ) - logger.info(f"Triggered workflow {workflow_id} in {repo_name} on {ref}") - except GithubException as e: - error_message = f"Failed to trigger workflow: {str(e)}" - logger.error(error_message) - raise + url = f"/repos/{self.org_name}/{repo_name}/actions/workflows/{workflow_id}/dispatches" + workflow_inputs = inputs if inputs is not None else {} + + self._request("POST", url, json={ + "ref": ref, + "inputs": workflow_inputs + }) + + logger.info(f"Triggered workflow {workflow_id} in {repo_name} on {ref}") def set_team_permission(self, repo_name: str, team_name: str, permission: str) -> None: """Set a team's permission on a repository. @@ -451,19 +414,11 @@ def set_team_permission(self, repo_name: str, team_name: str, permission: str) - repo_name: Name of the repository team_name: Name of the team permission: Permission level ('pull', 'push', 'admin', 'maintain', 'triage') - - Raises: - GithubException: If the operation fails """ - try: - repo = self.org.get_repo(repo_name) - team = self.org.get_team_by_slug(team_name) - team.update_team_repository(repo, permission) - logger.info(f"Set {team_name} permission on {repo_name} to {permission}") - except GithubException as e: - error_message = f"Failed to set team permission: {str(e)}" - logger.error(error_message) - raise + url = f"/orgs/{self.org_name}/teams/{team_name}/repos/{self.org_name}/{repo_name}" + self._request("PUT", url, json={"permission": permission}) + + logger.info(f"Set {team_name} permission on {repo_name} to {permission}") def update_repository_topics(self, repo_name: str, topics: List[str]) -> None: """Update the topics of a repository. @@ -471,15 +426,48 @@ def update_repository_topics(self, repo_name: str, topics: List[str]) -> None: Args: repo_name: Name of the repository topics: List of topics to set + """ + # GitHub API requires a special media type for repository topics + headers = {"Accept": "application/vnd.github.mercy-preview+json"} + url = f"/repos/{self.org_name}/{repo_name}/topics" + + self._request("PUT", url, json={"names": topics}, headers=headers) + + logger.info(f"Updated topics for {repo_name}: {topics}") + + def create_repository_from_template( + self, + template_repo_name: str, + new_repo_name: str, + private: bool = True, + description: Optional[str] = None, + topics: Optional[List[str]] = None + ) -> Dict[str, Any]: + """Create a new repository from a template. + + Args: + template_repo_name: Name of the template repository + new_repo_name: Name for the new repository + private: Whether the new repository should be private + description: Description for the new repository + topics: List of topics to add to the repository - Raises: - GithubException: If the operation fails + Returns: + The newly created repository """ - try: - repo = self.org.get_repo(repo_name) - repo.replace_topics(topics) - logger.info(f"Updated topics for {repo_name}: {topics}") - except GithubException as e: - error_message = f"Failed to update repository topics: {str(e)}" - logger.error(error_message) - raise + url = f"/repos/{self.org_name}/{template_repo_name}/generate" + + # Create repository from template + new_repo = self._request("POST", url, json={ + "name": new_repo_name, + "owner": self.org_name, + "description": description or f"Repository created from template: {template_repo_name}", + "private": private + }) + + # Add topics if provided + if topics: + self.update_repository_topics(new_repo_name, topics) + + logger.info(f"Created new repository: {new_repo_name} from template: {template_repo_name}") + return new_repo