#!/usr/bin/env python import base64 import random import requests from requests import Response from typing import Any, Dict, Optional from github import Github from default_params import ( DEFAULT_ACCESS_TOKEN, DEFAULT_DEVCONTAINER_JSON, DEFAULT_DOCKERFILE, DEFAULT_REPO_URL, DEFAULT_SAMPLE_SCRIPT, DEFAULT_USERNAME, ) class GithubRepoAccess: access_token: str gh: Github headers = {} def __init__(self, access_token) -> None: self.access_token = access_token self.gh = Github(self.access_token) self.headers["Authorization"] = self.access_token ## function that uses the github api for def clone_repository(self,repo_name: str) -> any: repo_details = self.gh.get_repo(repo_name) repo_url = repo_details.clone_url repo_details = requests.get(repo_url, headers=self.headers) ## getting principal branch for cloning the file branch = self.gh.get_repo(repo_name) clone_repository_url = ( f"https://api.github.com/repos/{repo_owner}/{repo_name}/tarball/{branch}" ) ## cloning the repository clone_repository_response = requests.get(clone_repository_url, headers=self.headers) def repo_exists(access_token: str, repo_name: str) -> bool: for repo in gh.get_user().get_repos(): if repo.name == repo_name: print("Repo exists!!") return True return False def fork_repository( username: str, repo_owner: str, repo_name: str, headers: dict[str, str] ) -> Optional[Dict[str, Any]]: fork_api_url = ( f"https://api.github.com/repos/{repo_owner}/{repo_name}/forks" ) fork_response = requests.post(fork_api_url, headers=self.headers) if fork_response.status_code == 202: print("Repository forked successfully.") return fork_response.json() else: print("Error forking the repository.") print("Status code:", fork_response.status_code) print("Error message:", fork_response.json()) return None def create_new_branch( username: str, repo_name: str, new_branch_name: str, headers: dict[str, str], ) -> None: api_base_url: str = f"https://api.github.com/repos/{username}/{repo_name}" branches_api_url: str = f"{api_base_url}/git/refs/heads" branches_response: Response = requests.get( branches_api_url, headers=self.headers ) branches = branches_response.json() main_branch_sha = None for branch in branches: if branch["ref"] == "refs/heads/main": main_branch_sha = branch["object"]["sha"] break if not main_branch_sha: print("Error: Couldn't find the main branch.") return new_branch_data: dict[str, Any] = { "ref": f"refs/heads/{new_branch_name}", "sha": main_branch_sha, } new_branch_response: Response = requests.post( branches_api_url, headers=self.headers, json=new_branch_data ) if new_branch_response.status_code == 201: print(f"New branch '{new_branch_name}' created successfully.") else: print("Error creating the new branch.") print("Status code:", new_branch_response.status_code) print("Error message:", new_branch_response.json()) def commit_files_to_branch( self, repo_owner: str, repo_name: str, new_branch_name: str, devcontainer_json_content: str, dockerfile_content: str, sample_script_content: str, headers: dict[str, str], ) -> None: api_base_url: str = ( f"https://api.github.com/repos/{repo_owner}/{repo_name}" ) # Get default branch and its commit SHA repo_info = requests.get(api_base_url, headers=self.headers).json() print(repo_info) default_branch = repo_info["default_branch"] default_branch_sha = requests.get( f"{api_base_url}/git/ref/heads/master", headers=self.headers ).json()["object"]["sha"] devcontainer_json_blob_sha = requests.post( f"{api_base_url}/git/blobs", headers=self.headers, json={ "content": base64.b64encode( devcontainer_json_content.encode() ).decode(), "encoding": "base64", }, ).json()["sha"] dockerfile_blob_sha = requests.post( f"{api_base_url}/git/blobs", headers=self.headers, json={ "content": base64.b64encode(dockerfile_content.encode()).decode(), "encoding": "base64", }, ).json()["sha"] sample_script_blob_sha = requests.post( f"{api_base_url}/git/blobs", headers=self.headers, json={ "content": base64.b64encode( sample_script_content.encode() ).decode(), "encoding": "base64", }, ).json()["sha"] # Get latest commit tree latest_commit_tree_sha = requests.get( f"{api_base_url}/git/commits/{default_branch_sha}", headers=self.headers ).json()["tree"]["sha"] print("Latest commit tree SHA:", latest_commit_tree_sha) # Create a new tree with the new blobs new_tree_response = requests.post( f"{api_base_url}/git/trees", headers=self.headers, json={ "base_tree": latest_commit_tree_sha, "tree": [ { "path": ".devcontainer/devcontainer.json", "mode": "100644", "type": "blob", "sha": devcontainer_json_blob_sha, }, { "path": ".devcontainer/Dockerfile", "mode": "100644", "type": "blob", "sha": dockerfile_blob_sha, }, { "path": "sample_script.py", "mode": "100644", "type": "blob", "sha": sample_script_blob_sha, }, ], }, ) if new_tree_response.status_code == 201: new_tree = new_tree_response.json() print("New tree created successfully.") print("New tree SHA:", new_tree["sha"]) else: print("Error creating the new tree.") print("Status code:", new_tree_response.status_code) print("Error message:", new_tree_response.json()) exit(1) # Create a new commit with the new tree new_commit_response = requests.post( f"{api_base_url}/git/commits", headers=self.headers, json={ "message": "Add devcontainer.json and Dockerfile", "tree": new_tree["sha"], "parents": [default_branch_sha], }, ) if new_commit_response.status_code == 201: new_commit = new_commit_response.json() print("New commit created successfully.") print("New commit SHA:", new_commit["sha"]) else: print("Error creating the new commit.") print("Status code:", new_commit_response.status_code) print("Error message:", new_commit_response.json()) exit(1) # Create new branch on the forked repository with the new commit SHA new_branch_ref = f"refs/heads/{new_branch_name}" create_branch_response = requests.post( f"{api_base_url}/git/refs", headers=self.headers, json={"ref": new_branch_ref, "sha": new_commit["sha"]}, ) if create_branch_response.status_code == 201: print( f"New branch '{new_branch_name}' created successfully on the forked repository with devcontainer.json and Dockerfile." ) else: print("Error creating the new branch on the forked repository.") print("Status code:", create_branch_response.status_code) print("Error message:", create_branch_response.json()) exit(1) def create_codespace( repo_owner: str, repo_name: str, new_branch_name: str, headers: dict[str, str], ) -> str: api_base_url: str = ( f"https://api.github.com/repos/{repo_owner}/{repo_name}/codespaces" ) create_codespace_payload = {"ref": new_branch_name} create_codespace_response = requests.post( api_base_url, headers=self.headers, json=create_codespace_payload ) if create_codespace_response.status_code == 201: print( "Codespace creation request is successful. Waiting for the Codespace to be created..." ) codespace = create_codespace_response.json() # Poll the Codespace's status until it becomes 'available' codespace_id = codespace["id"] print(codespace_id) print(codespace) codespace_status = codespace["state"] print(codespace_status) return codespace_id # while codespace_status != 'Available': # time.sleep(10) # codespace_response = requests.get(f'{api_base_url}/{codespace_id}', headers=self.headers) # codespace = codespace_response.json() # import ipdb # ipdb.set_trace() # codespace_status = codespace['state'] # print(f"Current Codespace status: {codespace_status}") # # print(f"Codespace is available! ID: {codespace_id}") else: print("Error creating the Codespace.") print("Status code:", create_codespace_response.status_code) print("Error message:", create_codespace_response.json()) def create_codespace_with_files( username: str, access_token: str, repo_url: str, docker_file: str, devcontainer_json: str, sample_script: str, ) -> str: # Extract repository owner and name from the repo URL repo_parts = repo_url.split("/") repo_owner = repo_parts[-2] repo_name = repo_parts[-1].replace(".git", "") # Configure headers for the GitHub API headers = { "Authorization": f"token {access_token}", "Accept": "application/vnd.github+json", "Content-Type": "application/json", } if not repo_exists(access_token, repo_name): # Fork the repository forked_repo = fork_repository(username, repo_owner, repo_name, headers) print("Forked!") # Create a new branch in the forked repository new_branch_name = "devcontainer-setup-" + str(random.randint(1, 1000)) # create_new_branch(username, repo_name, new_branch_name, headers) # Commit devcontainer.json, Dockerfile, and sample_script to the new branch commit_files_to_branch( username, repo_name, new_branch_name, devcontainer_json, docker_file, sample_script, headers, ) print("Branch created and committed files") # Create a new Codespace using the new branch codespace_id = create_codespace( username, repo_name, new_branch_name, headers ) return codespace_id if __name__ == "__main__": create_codespace_with_files( username=DEFAULT_USERNAME, access_token=DEFAULT_ACCESS_TOKEN, repo_url=DEFAULT_REPO_URL, docker_file=DEFAULT_DOCKERFILE, devcontainer_json=DEFAULT_DEVCONTAINER_JSON, sample_script=DEFAULT_SAMPLE_SCRIPT, )