From 84af37a27a97a31a8dfa0e9578b3a161b3ac3d37 Mon Sep 17 00:00:00 2001 From: Bently Date: Fri, 3 Jan 2025 15:02:53 +0000 Subject: [PATCH] refactor(blocks): Move some GitHub blocks to correct file (#9180) This moves my recently added blocks: ``GithubCreateFileBlock``, ``GithubUpdateFileBlock``, ``GithubCreateRepositoryBlock`` and ``GithubListStargazersBlock`` to the correct file ``github/repo.py`` as i placed them in the wrong file originally --- .../backend/blocks/github/pull_requests.py | 418 ------------------ .../backend/backend/blocks/github/repo.py | 417 +++++++++++++++++ 2 files changed, 417 insertions(+), 418 deletions(-) diff --git a/autogpt_platform/backend/backend/blocks/github/pull_requests.py b/autogpt_platform/backend/backend/blocks/github/pull_requests.py index f890ba0d5..e8fad2daa 100644 --- a/autogpt_platform/backend/backend/blocks/github/pull_requests.py +++ b/autogpt_platform/backend/backend/blocks/github/pull_requests.py @@ -1,4 +1,3 @@ -import base64 import re from typing_extensions import TypedDict @@ -513,420 +512,3 @@ def prepare_pr_api_url(pr_url: str, path: str) -> str: base_url, pr_number = match.groups() return f"{base_url}/pulls/{pr_number}/{path}" - - -class GithubCreateFileBlock(Block): - class Input(BlockSchema): - credentials: GithubCredentialsInput = GithubCredentialsField("repo") - repo_url: str = SchemaField( - description="URL of the GitHub repository", - placeholder="https://github.com/owner/repo", - ) - file_path: str = SchemaField( - description="Path where the file should be created", - placeholder="path/to/file.txt", - ) - content: str = SchemaField( - description="Content to write to the file", - placeholder="File content here", - ) - branch: str = SchemaField( - description="Branch where the file should be created", - default="main", - ) - commit_message: str = SchemaField( - description="Message for the commit", - default="Create new file", - ) - - class Output(BlockSchema): - url: str = SchemaField(description="URL of the created file") - sha: str = SchemaField(description="SHA of the commit") - error: str = SchemaField( - description="Error message if the file creation failed" - ) - - def __init__(self): - super().__init__( - id="8fd132ac-b917-428a-8159-d62893e8a3fe", - description="This block creates a new file in a GitHub repository.", - categories={BlockCategory.DEVELOPER_TOOLS}, - input_schema=GithubCreateFileBlock.Input, - output_schema=GithubCreateFileBlock.Output, - test_input={ - "repo_url": "https://github.com/owner/repo", - "file_path": "test/file.txt", - "content": "Test content", - "branch": "main", - "commit_message": "Create test file", - "credentials": TEST_CREDENTIALS_INPUT, - }, - test_credentials=TEST_CREDENTIALS, - test_output=[ - ("url", "https://github.com/owner/repo/blob/main/test/file.txt"), - ("sha", "abc123"), - ], - test_mock={ - "create_file": lambda *args, **kwargs: ( - "https://github.com/owner/repo/blob/main/test/file.txt", - "abc123", - ) - }, - ) - - @staticmethod - def create_file( - credentials: GithubCredentials, - repo_url: str, - file_path: str, - content: str, - branch: str, - commit_message: str, - ) -> tuple[str, str]: - api = get_api(credentials) - # Convert content to base64 - content_bytes = content.encode("utf-8") - content_base64 = base64.b64encode(content_bytes).decode("utf-8") - - # Create the file using the GitHub API - contents_url = f"{repo_url}/contents/{file_path}" - data = { - "message": commit_message, - "content": content_base64, - "branch": branch, - } - response = api.put(contents_url, json=data) - result = response.json() - - return result["content"]["html_url"], result["commit"]["sha"] - - def run( - self, - input_data: Input, - *, - credentials: GithubCredentials, - **kwargs, - ) -> BlockOutput: - try: - url, sha = self.create_file( - credentials, - input_data.repo_url, - input_data.file_path, - input_data.content, - input_data.branch, - input_data.commit_message, - ) - yield "url", url - yield "sha", sha - except Exception as e: - yield "error", str(e) - - -class GithubUpdateFileBlock(Block): - class Input(BlockSchema): - credentials: GithubCredentialsInput = GithubCredentialsField("repo") - repo_url: str = SchemaField( - description="URL of the GitHub repository", - placeholder="https://github.com/owner/repo", - ) - file_path: str = SchemaField( - description="Path to the file to update", - placeholder="path/to/file.txt", - ) - content: str = SchemaField( - description="New content for the file", - placeholder="Updated content here", - ) - branch: str = SchemaField( - description="Branch containing the file", - default="main", - ) - commit_message: str = SchemaField( - description="Message for the commit", - default="Update file", - ) - - class Output(BlockSchema): - url: str = SchemaField(description="URL of the updated file") - sha: str = SchemaField(description="SHA of the commit") - error: str = SchemaField(description="Error message if the file update failed") - - def __init__(self): - super().__init__( - id="30be12a4-57cb-4aa4-baf5-fcc68d136076", - description="This block updates an existing file in a GitHub repository.", - categories={BlockCategory.DEVELOPER_TOOLS}, - input_schema=GithubUpdateFileBlock.Input, - output_schema=GithubUpdateFileBlock.Output, - test_input={ - "repo_url": "https://github.com/owner/repo", - "file_path": "test/file.txt", - "content": "Updated content", - "branch": "main", - "commit_message": "Update test file", - "credentials": TEST_CREDENTIALS_INPUT, - }, - test_credentials=TEST_CREDENTIALS, - test_output=[ - ("url", "https://github.com/owner/repo/blob/main/test/file.txt"), - ("sha", "def456"), - ], - test_mock={ - "update_file": lambda *args, **kwargs: ( - "https://github.com/owner/repo/blob/main/test/file.txt", - "def456", - ) - }, - ) - - @staticmethod - def update_file( - credentials: GithubCredentials, - repo_url: str, - file_path: str, - content: str, - branch: str, - commit_message: str, - ) -> tuple[str, str]: - api = get_api(credentials) - - # First get the current file to get its SHA - contents_url = f"{repo_url}/contents/{file_path}" - params = {"ref": branch} - response = api.get(contents_url, params=params) - current_file = response.json() - - # Convert new content to base64 - content_bytes = content.encode("utf-8") - content_base64 = base64.b64encode(content_bytes).decode("utf-8") - - # Update the file - data = { - "message": commit_message, - "content": content_base64, - "sha": current_file["sha"], - "branch": branch, - } - response = api.put(contents_url, json=data) - result = response.json() - - return result["content"]["html_url"], result["commit"]["sha"] - - def run( - self, - input_data: Input, - *, - credentials: GithubCredentials, - **kwargs, - ) -> BlockOutput: - try: - url, sha = self.update_file( - credentials, - input_data.repo_url, - input_data.file_path, - input_data.content, - input_data.branch, - input_data.commit_message, - ) - yield "url", url - yield "sha", sha - except Exception as e: - yield "error", str(e) - - -class GithubCreateRepositoryBlock(Block): - class Input(BlockSchema): - credentials: GithubCredentialsInput = GithubCredentialsField("repo") - name: str = SchemaField( - description="Name of the repository to create", - placeholder="my-new-repo", - ) - description: str = SchemaField( - description="Description of the repository", - placeholder="A description of the repository", - default="", - ) - private: bool = SchemaField( - description="Whether the repository should be private", - default=False, - ) - auto_init: bool = SchemaField( - description="Whether to initialize the repository with a README", - default=True, - ) - gitignore_template: str = SchemaField( - description="Git ignore template to use (e.g., Python, Node, Java)", - default="", - ) - - class Output(BlockSchema): - url: str = SchemaField(description="URL of the created repository") - clone_url: str = SchemaField(description="Git clone URL of the repository") - error: str = SchemaField( - description="Error message if the repository creation failed" - ) - - def __init__(self): - super().__init__( - id="029ec3b8-1cfd-46d3-b6aa-28e4a706efd1", - description="This block creates a new GitHub repository.", - categories={BlockCategory.DEVELOPER_TOOLS}, - input_schema=GithubCreateRepositoryBlock.Input, - output_schema=GithubCreateRepositoryBlock.Output, - test_input={ - "name": "test-repo", - "description": "A test repository", - "private": False, - "auto_init": True, - "gitignore_template": "Python", - "credentials": TEST_CREDENTIALS_INPUT, - }, - test_credentials=TEST_CREDENTIALS, - test_output=[ - ("url", "https://github.com/owner/test-repo"), - ("clone_url", "https://github.com/owner/test-repo.git"), - ], - test_mock={ - "create_repository": lambda *args, **kwargs: ( - "https://github.com/owner/test-repo", - "https://github.com/owner/test-repo.git", - ) - }, - ) - - @staticmethod - def create_repository( - credentials: GithubCredentials, - name: str, - description: str, - private: bool, - auto_init: bool, - gitignore_template: str, - ) -> tuple[str, str]: - api = get_api(credentials, convert_urls=False) # Disable URL conversion - data = { - "name": name, - "description": description, - "private": private, - "auto_init": auto_init, - } - - if gitignore_template: - data["gitignore_template"] = gitignore_template - - # Create repository using the user endpoint - response = api.post("https://api.github.com/user/repos", json=data) - result = response.json() - - return result["html_url"], result["clone_url"] - - def run( - self, - input_data: Input, - *, - credentials: GithubCredentials, - **kwargs, - ) -> BlockOutput: - try: - url, clone_url = self.create_repository( - credentials, - input_data.name, - input_data.description, - input_data.private, - input_data.auto_init, - input_data.gitignore_template, - ) - yield "url", url - yield "clone_url", clone_url - except Exception as e: - yield "error", str(e) - - -class GithubListStargazersBlock(Block): - class Input(BlockSchema): - credentials: GithubCredentialsInput = GithubCredentialsField("repo") - repo_url: str = SchemaField( - description="URL of the GitHub repository", - placeholder="https://github.com/owner/repo", - ) - - class Output(BlockSchema): - class StargazerItem(TypedDict): - username: str - url: str - - stargazer: StargazerItem = SchemaField( - title="Stargazer", - description="Stargazers with their username and profile URL", - ) - error: str = SchemaField( - description="Error message if listing stargazers failed" - ) - - def __init__(self): - super().__init__( - id="a4b9c2d1-e5f6-4g7h-8i9j-0k1l2m3n4o5p", # Generated unique UUID - description="This block lists all users who have starred a specified GitHub repository.", - categories={BlockCategory.DEVELOPER_TOOLS}, - input_schema=GithubListStargazersBlock.Input, - output_schema=GithubListStargazersBlock.Output, - test_input={ - "repo_url": "https://github.com/owner/repo", - "credentials": TEST_CREDENTIALS_INPUT, - }, - test_credentials=TEST_CREDENTIALS, - test_output=[ - ( - "stargazer", - { - "username": "octocat", - "url": "https://github.com/octocat", - }, - ) - ], - test_mock={ - "list_stargazers": lambda *args, **kwargs: [ - { - "username": "octocat", - "url": "https://github.com/octocat", - } - ] - }, - ) - - @staticmethod - def list_stargazers( - credentials: GithubCredentials, repo_url: str - ) -> list[Output.StargazerItem]: - api = get_api(credentials) - # Add /stargazers to the repo URL to get stargazers endpoint - stargazers_url = f"{repo_url}/stargazers" - # Set accept header to get starred_at timestamp - headers = {"Accept": "application/vnd.github.star+json"} - response = api.get(stargazers_url, headers=headers) - data = response.json() - - stargazers: list[GithubListStargazersBlock.Output.StargazerItem] = [ - { - "username": stargazer["login"], - "url": stargazer["html_url"], - } - for stargazer in data - ] - return stargazers - - def run( - self, - input_data: Input, - *, - credentials: GithubCredentials, - **kwargs, - ) -> BlockOutput: - try: - stargazers = self.list_stargazers( - credentials, - input_data.repo_url, - ) - yield from (("stargazer", stargazer) for stargazer in stargazers) - except Exception as e: - yield "error", str(e) diff --git a/autogpt_platform/backend/backend/blocks/github/repo.py b/autogpt_platform/backend/backend/blocks/github/repo.py index 7e2521181..82bef9475 100644 --- a/autogpt_platform/backend/backend/blocks/github/repo.py +++ b/autogpt_platform/backend/backend/blocks/github/repo.py @@ -699,3 +699,420 @@ class GithubDeleteBranchBlock(Block): input_data.branch, ) yield "status", status + + +class GithubCreateFileBlock(Block): + class Input(BlockSchema): + credentials: GithubCredentialsInput = GithubCredentialsField("repo") + repo_url: str = SchemaField( + description="URL of the GitHub repository", + placeholder="https://github.com/owner/repo", + ) + file_path: str = SchemaField( + description="Path where the file should be created", + placeholder="path/to/file.txt", + ) + content: str = SchemaField( + description="Content to write to the file", + placeholder="File content here", + ) + branch: str = SchemaField( + description="Branch where the file should be created", + default="main", + ) + commit_message: str = SchemaField( + description="Message for the commit", + default="Create new file", + ) + + class Output(BlockSchema): + url: str = SchemaField(description="URL of the created file") + sha: str = SchemaField(description="SHA of the commit") + error: str = SchemaField( + description="Error message if the file creation failed" + ) + + def __init__(self): + super().__init__( + id="8fd132ac-b917-428a-8159-d62893e8a3fe", + description="This block creates a new file in a GitHub repository.", + categories={BlockCategory.DEVELOPER_TOOLS}, + input_schema=GithubCreateFileBlock.Input, + output_schema=GithubCreateFileBlock.Output, + test_input={ + "repo_url": "https://github.com/owner/repo", + "file_path": "test/file.txt", + "content": "Test content", + "branch": "main", + "commit_message": "Create test file", + "credentials": TEST_CREDENTIALS_INPUT, + }, + test_credentials=TEST_CREDENTIALS, + test_output=[ + ("url", "https://github.com/owner/repo/blob/main/test/file.txt"), + ("sha", "abc123"), + ], + test_mock={ + "create_file": lambda *args, **kwargs: ( + "https://github.com/owner/repo/blob/main/test/file.txt", + "abc123", + ) + }, + ) + + @staticmethod + def create_file( + credentials: GithubCredentials, + repo_url: str, + file_path: str, + content: str, + branch: str, + commit_message: str, + ) -> tuple[str, str]: + api = get_api(credentials) + # Convert content to base64 + content_bytes = content.encode("utf-8") + content_base64 = base64.b64encode(content_bytes).decode("utf-8") + + # Create the file using the GitHub API + contents_url = f"{repo_url}/contents/{file_path}" + data = { + "message": commit_message, + "content": content_base64, + "branch": branch, + } + response = api.put(contents_url, json=data) + result = response.json() + + return result["content"]["html_url"], result["commit"]["sha"] + + def run( + self, + input_data: Input, + *, + credentials: GithubCredentials, + **kwargs, + ) -> BlockOutput: + try: + url, sha = self.create_file( + credentials, + input_data.repo_url, + input_data.file_path, + input_data.content, + input_data.branch, + input_data.commit_message, + ) + yield "url", url + yield "sha", sha + except Exception as e: + yield "error", str(e) + + +class GithubUpdateFileBlock(Block): + class Input(BlockSchema): + credentials: GithubCredentialsInput = GithubCredentialsField("repo") + repo_url: str = SchemaField( + description="URL of the GitHub repository", + placeholder="https://github.com/owner/repo", + ) + file_path: str = SchemaField( + description="Path to the file to update", + placeholder="path/to/file.txt", + ) + content: str = SchemaField( + description="New content for the file", + placeholder="Updated content here", + ) + branch: str = SchemaField( + description="Branch containing the file", + default="main", + ) + commit_message: str = SchemaField( + description="Message for the commit", + default="Update file", + ) + + class Output(BlockSchema): + url: str = SchemaField(description="URL of the updated file") + sha: str = SchemaField(description="SHA of the commit") + error: str = SchemaField(description="Error message if the file update failed") + + def __init__(self): + super().__init__( + id="30be12a4-57cb-4aa4-baf5-fcc68d136076", + description="This block updates an existing file in a GitHub repository.", + categories={BlockCategory.DEVELOPER_TOOLS}, + input_schema=GithubUpdateFileBlock.Input, + output_schema=GithubUpdateFileBlock.Output, + test_input={ + "repo_url": "https://github.com/owner/repo", + "file_path": "test/file.txt", + "content": "Updated content", + "branch": "main", + "commit_message": "Update test file", + "credentials": TEST_CREDENTIALS_INPUT, + }, + test_credentials=TEST_CREDENTIALS, + test_output=[ + ("url", "https://github.com/owner/repo/blob/main/test/file.txt"), + ("sha", "def456"), + ], + test_mock={ + "update_file": lambda *args, **kwargs: ( + "https://github.com/owner/repo/blob/main/test/file.txt", + "def456", + ) + }, + ) + + @staticmethod + def update_file( + credentials: GithubCredentials, + repo_url: str, + file_path: str, + content: str, + branch: str, + commit_message: str, + ) -> tuple[str, str]: + api = get_api(credentials) + + # First get the current file to get its SHA + contents_url = f"{repo_url}/contents/{file_path}" + params = {"ref": branch} + response = api.get(contents_url, params=params) + current_file = response.json() + + # Convert new content to base64 + content_bytes = content.encode("utf-8") + content_base64 = base64.b64encode(content_bytes).decode("utf-8") + + # Update the file + data = { + "message": commit_message, + "content": content_base64, + "sha": current_file["sha"], + "branch": branch, + } + response = api.put(contents_url, json=data) + result = response.json() + + return result["content"]["html_url"], result["commit"]["sha"] + + def run( + self, + input_data: Input, + *, + credentials: GithubCredentials, + **kwargs, + ) -> BlockOutput: + try: + url, sha = self.update_file( + credentials, + input_data.repo_url, + input_data.file_path, + input_data.content, + input_data.branch, + input_data.commit_message, + ) + yield "url", url + yield "sha", sha + except Exception as e: + yield "error", str(e) + + +class GithubCreateRepositoryBlock(Block): + class Input(BlockSchema): + credentials: GithubCredentialsInput = GithubCredentialsField("repo") + name: str = SchemaField( + description="Name of the repository to create", + placeholder="my-new-repo", + ) + description: str = SchemaField( + description="Description of the repository", + placeholder="A description of the repository", + default="", + ) + private: bool = SchemaField( + description="Whether the repository should be private", + default=False, + ) + auto_init: bool = SchemaField( + description="Whether to initialize the repository with a README", + default=True, + ) + gitignore_template: str = SchemaField( + description="Git ignore template to use (e.g., Python, Node, Java)", + default="", + ) + + class Output(BlockSchema): + url: str = SchemaField(description="URL of the created repository") + clone_url: str = SchemaField(description="Git clone URL of the repository") + error: str = SchemaField( + description="Error message if the repository creation failed" + ) + + def __init__(self): + super().__init__( + id="029ec3b8-1cfd-46d3-b6aa-28e4a706efd1", + description="This block creates a new GitHub repository.", + categories={BlockCategory.DEVELOPER_TOOLS}, + input_schema=GithubCreateRepositoryBlock.Input, + output_schema=GithubCreateRepositoryBlock.Output, + test_input={ + "name": "test-repo", + "description": "A test repository", + "private": False, + "auto_init": True, + "gitignore_template": "Python", + "credentials": TEST_CREDENTIALS_INPUT, + }, + test_credentials=TEST_CREDENTIALS, + test_output=[ + ("url", "https://github.com/owner/test-repo"), + ("clone_url", "https://github.com/owner/test-repo.git"), + ], + test_mock={ + "create_repository": lambda *args, **kwargs: ( + "https://github.com/owner/test-repo", + "https://github.com/owner/test-repo.git", + ) + }, + ) + + @staticmethod + def create_repository( + credentials: GithubCredentials, + name: str, + description: str, + private: bool, + auto_init: bool, + gitignore_template: str, + ) -> tuple[str, str]: + api = get_api(credentials, convert_urls=False) # Disable URL conversion + data = { + "name": name, + "description": description, + "private": private, + "auto_init": auto_init, + } + + if gitignore_template: + data["gitignore_template"] = gitignore_template + + # Create repository using the user endpoint + response = api.post("https://api.github.com/user/repos", json=data) + result = response.json() + + return result["html_url"], result["clone_url"] + + def run( + self, + input_data: Input, + *, + credentials: GithubCredentials, + **kwargs, + ) -> BlockOutput: + try: + url, clone_url = self.create_repository( + credentials, + input_data.name, + input_data.description, + input_data.private, + input_data.auto_init, + input_data.gitignore_template, + ) + yield "url", url + yield "clone_url", clone_url + except Exception as e: + yield "error", str(e) + + +class GithubListStargazersBlock(Block): + class Input(BlockSchema): + credentials: GithubCredentialsInput = GithubCredentialsField("repo") + repo_url: str = SchemaField( + description="URL of the GitHub repository", + placeholder="https://github.com/owner/repo", + ) + + class Output(BlockSchema): + class StargazerItem(TypedDict): + username: str + url: str + + stargazer: StargazerItem = SchemaField( + title="Stargazer", + description="Stargazers with their username and profile URL", + ) + error: str = SchemaField( + description="Error message if listing stargazers failed" + ) + + def __init__(self): + super().__init__( + id="a4b9c2d1-e5f6-4g7h-8i9j-0k1l2m3n4o5p", # Generated unique UUID + description="This block lists all users who have starred a specified GitHub repository.", + categories={BlockCategory.DEVELOPER_TOOLS}, + input_schema=GithubListStargazersBlock.Input, + output_schema=GithubListStargazersBlock.Output, + test_input={ + "repo_url": "https://github.com/owner/repo", + "credentials": TEST_CREDENTIALS_INPUT, + }, + test_credentials=TEST_CREDENTIALS, + test_output=[ + ( + "stargazer", + { + "username": "octocat", + "url": "https://github.com/octocat", + }, + ) + ], + test_mock={ + "list_stargazers": lambda *args, **kwargs: [ + { + "username": "octocat", + "url": "https://github.com/octocat", + } + ] + }, + ) + + @staticmethod + def list_stargazers( + credentials: GithubCredentials, repo_url: str + ) -> list[Output.StargazerItem]: + api = get_api(credentials) + # Add /stargazers to the repo URL to get stargazers endpoint + stargazers_url = f"{repo_url}/stargazers" + # Set accept header to get starred_at timestamp + headers = {"Accept": "application/vnd.github.star+json"} + response = api.get(stargazers_url, headers=headers) + data = response.json() + + stargazers: list[GithubListStargazersBlock.Output.StargazerItem] = [ + { + "username": stargazer["login"], + "url": stargazer["html_url"], + } + for stargazer in data + ] + return stargazers + + def run( + self, + input_data: Input, + *, + credentials: GithubCredentials, + **kwargs, + ) -> BlockOutput: + try: + stargazers = self.list_stargazers( + credentials, + input_data.repo_url, + ) + yield from (("stargazer", stargazer) for stargazer in stargazers) + except Exception as e: + yield "error", str(e)