mirror of
https://github.com/johndoe6345789/AutoMetabuilder.git
synced 2026-04-24 13:54:59 +00:00
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
"""
|
|
GitHub integration module.
|
|
"""
|
|
import os
|
|
from github import Github
|
|
from github.Issue import Issue
|
|
from github.PullRequest import PullRequest
|
|
from tenacity import retry, stop_after_attempt, wait_exponential
|
|
|
|
from .. import load_messages
|
|
|
|
|
|
class GitHubIntegration:
|
|
"""Class to handle GitHub interactions."""
|
|
|
|
def __init__(self, token: str, repo_name: str):
|
|
self.github = Github(token)
|
|
self.repo = self.github.get_repo(repo_name)
|
|
|
|
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
|
|
def get_open_issues(self):
|
|
"""Get open issues from the repository."""
|
|
return self.repo.get_issues(state='open')
|
|
|
|
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
|
|
def get_issue(self, issue_number: int) -> Issue:
|
|
"""Get a specific issue by number."""
|
|
return self.repo.get_issue(number=issue_number)
|
|
|
|
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
|
|
def create_branch(self, branch_name: str, base_branch: str = "main"):
|
|
"""Create a new branch from a base branch."""
|
|
base_ref = self.repo.get_git_ref(f"heads/{base_branch}")
|
|
self.repo.create_git_ref(
|
|
ref=f"refs/heads/{branch_name}", sha=base_ref.object.sha
|
|
)
|
|
|
|
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
|
|
def create_pull_request(
|
|
self,
|
|
title: str,
|
|
body: str,
|
|
head_branch: str,
|
|
base_branch: str = "main",
|
|
) -> PullRequest:
|
|
"""Create a new pull request."""
|
|
return self.repo.create_pull(
|
|
title=title, body=body, head=head_branch, base=base_branch
|
|
)
|
|
|
|
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
|
|
def get_pull_requests(self, state: str = "open"):
|
|
"""Get pull requests from the repository."""
|
|
return self.repo.get_pulls(state=state)
|
|
|
|
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
|
|
def get_pull_request_comments(self, pr_number: int):
|
|
"""Get comments from a specific pull request."""
|
|
pr = self.repo.get_pull(pr_number)
|
|
return pr.get_issue_comments()
|
|
|
|
|
|
def get_repo_name_from_env() -> str:
|
|
"""Retrieve repository name from environment variable."""
|
|
# Try to get from environment variable
|
|
repo_name = os.environ.get("GITHUB_REPOSITORY")
|
|
if not repo_name:
|
|
# Fallback or error
|
|
msgs = load_messages()
|
|
raise ValueError(msgs["error_github_repo_missing"])
|
|
return repo_name
|