# /mnt/e/genesis-system/core/integrations/github_integration.py import os import logging import requests from typing import Optional, Dict, Any, List # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class GitHubIntegration: """ A class to interact with the GitHub API for repository management. """ def __init__(self, token: str, owner: str): """ Initializes the GitHubIntegration class. Args: token (str): GitHub personal access token. owner (str): The owner of the GitHub repository (username or organization). """ self.token = token self.owner = owner self.api_url = "https://api.github.com" self.headers = { "Authorization": f"token {self.token}", "Accept": "application/vnd.github.v3+json" } def _make_request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]: """ Makes a request to the GitHub API. Args: method (str): HTTP method (e.g., "GET", "POST", "PATCH", "DELETE"). endpoint (str): The API endpoint. data (Optional[Dict[str, Any]]): Data to send in the request body (for POST, PATCH). Returns: Optional[Dict[str, Any]]: The JSON response from the API, or None if an error occurred. """ url = f"{self.api_url}{endpoint}" logging.info(f"Making {method} request to {url} with data: {data}") try: response = requests.request(method, url, headers=self.headers, json=data) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) if response.status_code != 204: #No Content status, means no json is returned return response.json() else: return None # 204 No Content doesn't return JSON except requests.exceptions.RequestException as e: logging.error(f"Request failed: {e}") return None def create_repository(self, repo_name: str, description: str, private: bool = False) -> Optional[Dict[str, Any]]: """ Creates a new GitHub repository. Args: repo_name (str): The name of the repository. description (str): A description for the repository. private (bool): Whether the repository should be private. Defaults to False. Returns: Optional[Dict[str, Any]]: The JSON response from the API, or None if an error occurred. """ endpoint = "/user/repos" # or /orgs/{org}/repos for organization repos data = { "name": repo_name, "description": description, "private": private } return self._make_request("POST", endpoint, data) def get_repository(self, repo_name: str) -> Optional[Dict[str, Any]]: """ Gets information about a specific repository. Args: repo_name (str): The name of the repository. Returns: Optional[Dict[str, Any]]: The JSON response from the API, or None if an error occurred. """ endpoint = f"/repos/{self.owner}/{repo_name}" return self._make_request("GET", endpoint) def update_repository(self, repo_name: str, description: Optional[str] = None, private: Optional[bool] = None) -> Optional[Dict[str, Any]]: """ Updates information about a specific repository. Args: repo_name (str): The name of the repository. description (Optional[str]): A new description for the repository. private (Optional[bool]): Whether the repository should be private. Returns: Optional[Dict[str, Any]]: The JSON response from the API, or None if an error occurred. """ endpoint = f"/repos/{self.owner}/{repo_name}" data = {} if description is not None: data["description"] = description if private is not None: data["private"] = private if not data: logging.warning("No update parameters provided.") return None return self._make_request("PATCH", endpoint, data) def delete_repository(self, repo_name: str) -> bool: """ Deletes a repository. Args: repo_name (str): The name of the repository. Returns: bool: True if the repository was successfully deleted, False otherwise. """ endpoint = f"/repos/{self.owner}/{repo_name}" response = self._make_request("DELETE", endpoint) return response is None # DELETE returns 204 No Content on success. _make_request returns None for 204 def list_repositories(self) -> Optional[List[Dict[str, Any]]]: """ Lists all repositories for the authenticated user. Returns: Optional[List[Dict[str, Any]]]: A list of repository dictionaries, or None if an error occurred. """ endpoint = "/user/repos" return self._make_request("GET", endpoint) if __name__ == '__main__': # Example Usage (replace with your actual token and owner) GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN") # Store your token securely (e.g., environment variable) GITHUB_OWNER = "your-github-username" # Replace with your GitHub username or organization name if not GITHUB_TOKEN: print("Error: GITHUB_TOKEN environment variable not set.") exit(1) github = GitHubIntegration(GITHUB_TOKEN, GITHUB_OWNER) # Example: Create a repository new_repo_name = "test-repo-genesis" repo_description = "A test repository created using the GitHub API integration." create_result = github.create_repository(new_repo_name, repo_description) if create_result: print(f"Repository '{new_repo_name}' created successfully: {create_result}") else: print(f"Failed to create repository '{new_repo_name}'.") # Example: Get repository details repo_details = github.get_repository(new_repo_name) if repo_details: print(f"Repository details: {repo_details}") else: print(f"Failed to get repository details for '{new_repo_name}'.") #Example: Update repository details updated_description = "Updated description for the test repo." update_result = github.update_repository(new_repo_name, description=updated_description) if update_result: print(f"Repository '{new_repo_name}' updated successfully: {update_result}") else: print(f"Failed to update repository '{new_repo_name}'.") # Example: List repositories repos = github.list_repositories() if repos: print("Repositories:") for repo in repos: print(f"- {repo['name']}") else: print("Failed to list repositories.") # Example: Delete the repository (cleanup) delete_success = github.delete_repository(new_repo_name) if delete_success: print(f"Repository '{new_repo_name}' deleted successfully.") else: print(f"Failed to delete repository '{new_repo_name}'.")