apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-06-08 15:05:17 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'feature/git/pr_issue' into 'mgx_ops'

Browse source 
feat: +git pr/issue

See merge request pub/MetaGPT!54
This commit is contained in:
洪思睿 2024-04-30 02:42:55 +00:00
commit ec86035d2e
3 changed files with 387 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

148
metagpt/tools/libs/git.py
View file

@ -3,6 +3,10 @@
from __future__ import annotations
from pathlib import Path
from typing import Optional
from github.Issue import Issue
from github.PullRequest import PullRequest
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.git_repository import GitRepository
@ -63,3 +67,147 @@ async def git_checkout(repo_dir: str | Path, commit_id: str):
if not repo.is_valid:
ValueError(f"Invalid git root: {repo_dir}")
await repo.checkout(commit_id)
@register_tool(tags=["git"])
async def create_pull_request(
access_token: str,
base: str,
head: str,
base_repo_name: str,
head_repo_name: Optional[str] = None,
title: Optional[str] = None,
body: Optional[str] = None,
) -> PullRequest:
"""
Creates a pull request in a Git repository.
Args:
access_token (str): The access token for authentication.
base (str): The name of the base branch of the pull request (e.g., 'main', 'master').
head (str): The name of the head branch of the pull request (e.g., 'feature-branch').
base_repo_name (str): The full repository name (user/repo) where the pull request will be created.
head_repo_name (Optional[str], optional): The full repository name (user/repo) where the pull request will merge from. Defaults to None.
title (Optional[str]): The title of the pull request.
body (Optional[str]): The body of the pull request.
Returns:
PullRequest: The created pull request object.
Raises:
ValueError: If `access_token` is invalid. Visit: "https://github.com/settings/tokens"
Any exceptions that might occur during the pull request creation process.
Note:
This function is intended to be used in an asynchronous context (with `await`).
Example:
>>> # Merge Request
>>> repo_name = "user/repo" # "user/repo" for example: "https://github.com/user/repo.git"
>>> base = "master" # branch that merge to
>>> head = "feature/new_feature" # branch that merge from
>>> title = "Implement new feature"
>>> body = "This pull request adds functionality X, Y, and Z."
>>> pull_request = await create_pull_request(
repo_name=repo_name,
base=base,
head=head,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> print(pull_request)
PullRequest(title="Implement new feature", number=26)
>>> # Pull Request
>>> base_repo_name = "user1/repo1" # for example: "user1/repo1" from "https://github.com/user1/repo1.git"
>>> head_repo_name = "user2/repo2" # for example: "user2/repo2" from "https://github.com/user2/repo2.git"
>>> base = "master" # branch that merge to
>>> head = "feature/new_feature" # branch that merge from
>>> title = "Implement new feature"
>>> body = "This pull request adds functionality X, Y, and Z."
>>> pull_request = await create_pull_request(
base_repo_name=base_repo_name,
head_repo_name=head_repo_name,
base=base,
head=head,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> print(pull_request)
PullRequest(title="Implement new feature", number=26)
"""
return await GitRepository.create_pull(
base_repo_name=base_repo_name,
head_repo_name=head_repo_name,
base=base,
head=head,
title=title,
body=body,
access_token=access_token,
)
@register_tool(tags=["git"])
async def create_issue(
access_token: str,
repo_name: str,
title: str,
body: Optional[str] = None,
assignee: Optional[str] = None,
labels: Optional[list[str]] = None,
) -> Issue:
"""
Creates an issue in the specified repository.
Args:
access_token (str): The access token for authentication.
Visit `https://github.com/settings/tokens` to obtain a personal access token.
For more authentication options, visit: `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`
repo_name (str): The full repository name (user/repo) where the issue will be created.
title (str): The title of the issue.
body (Optional[str], optional): The body of the issue. Defaults to None.
assignee (Optional[str], optional): The username of the assignee for the issue. Defaults to None.
labels (Optional[list[str]], optional): A list of label names to associate with the issue. Defaults to None.
Returns:
Issue: The created issue object.
Example:
>>> # Create an issue with title and body
>>> repo_name = "username/repository"
>>> title = "Bug Report"
>>> body = "I found a bug in the application."
>>> issue = await create_issue(
repo_name=repo_name,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> print(issue)
Issue(title="Bug Report", number=26)
>>> # Create an issue with title, body, assignee, and labels
>>> repo_name = "username/repository"
>>> title = "Bug Report"
>>> body = "I found a bug in the application."
>>> assignee = "john_doe"
>>> labels = ["enhancement", "help wanted"]
>>> issue = await create_issue(
repo_name=repo_name,
title=title,
body=body,
assignee=assigee,
labels=labels,
access_token=get_env("git_access_token")
)
>>> print(issue)
Issue(title="Bug Report", number=26)
"""
return await GitRepository.create_issue(
repo_name=repo_name, title=title, body=body, assignee=assignee, labels=labels, access_token=access_token
)

169
metagpt/utils/git_repository.py
View file

@ -8,14 +8,22 @@
"""
from __future__ import annotations
import re
import shutil
import uuid
from enum import Enum
from pathlib import Path
from typing import Dict, List
from typing import Dict, List, Optional
from git.repo import Repo
from git.repo.fun import is_git_dir
from github import Auth, Github
from github.GithubObject import NotSet
from github.Issue import Issue
from github.Label import Label
from github.Milestone import Milestone
from github.NamedUser import NamedUser
from github.PullRequest import PullRequest
from gitignore_parser import parse_gitignore
from tenacity import retry, stop_after_attempt, wait_random_exponential
@ -35,6 +43,12 @@ class ChangeType(Enum):
UNTRACTED = "U" # File is untracked (not added to version control)
class RateLimitError(Exception):
def __init__(self, message="Rate limit exceeded"):
self.message = message
super().__init__(self.message)
class GitRepository:
"""A class representing a Git repository.
@ -322,3 +336,156 @@ class GitRepository:
def log(self) -> str:
"""Return git log"""
return self._repository.git.log()
@staticmethod
async def create_pull(
base: str,
head: str,
base_repo_name: str,
head_repo_name: Optional[str] = None,
*,
title: Optional[str] = None,
body: Optional[str] = None,
maintainer_can_modify: Optional[bool] = None,
draft: Optional[bool] = None,
issue: Optional[Issue] = None,
access_token: Optional[str] = None,
auth: Optional[Auth] = None,
) -> PullRequest:
"""
Creates a pull request in the specified repository.
Args:
base (str): The name of the base branch.
head (str): The name of the head branch.
base_repo_name (str): The full repository name (user/repo) where the pull request will be created.
head_repo_name (Optional[str], optional): The full repository name (user/repo) where the pull request will merge from. Defaults to None.
title (Optional[str], optional): The title of the pull request. Defaults to None.
body (Optional[str], optional): The body of the pull request. Defaults to None.
maintainer_can_modify (Optional[bool], optional): Whether maintainers can modify the pull request. Defaults to None.
draft (Optional[bool], optional): Whether the pull request is a draft. Defaults to None.
issue (Optional[Issue], optional): The issue linked to the pull request. Defaults to None.
access_token (Optional[str], optional): The access token for authentication. Defaults to None. Visit `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`, `https://github.com/PyGithub/PyGithub/blob/main/doc/examples/Authentication.rst`.
auth (Optional[Auth], optional): The authentication method. Defaults to None. Visit `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`
Returns:
PullRequest: The created pull request object.
"""
title = title or NotSet
body = body or NotSet
maintainer_can_modify = maintainer_can_modify or NotSet
draft = draft or NotSet
issue = issue or NotSet
if not auth and not access_token:
raise ValueError('`access_token` is invalid. Visit: "https://github.com/settings/tokens"')
auth = auth or Auth.Token(access_token)
g = Github(auth=auth)
base_repo = g.get_repo(base_repo_name)
head_repo = g.get_repo(head_repo_name) if head_repo_name and head_repo_name != base_repo_name else None
x_ratelimit_remaining = base_repo.raw_headers.get("x-ratelimit-remaining")
if (
x_ratelimit_remaining
and bool(re.match(r"^-?\d+$", x_ratelimit_remaining))
and int(x_ratelimit_remaining) <= 0
):
raise RateLimitError()
if not head_repo:
pr = base_repo.create_pull(
base=base,
head=head,
title=title,
body=body,
maintainer_can_modify=maintainer_can_modify,
draft=draft,
issue=issue,
)
else:
head_branch = base_repo.get_branch(base)
base_branch = head_repo.get_branch(head)
pr = base_repo.create_pull(
base=base_branch.name,
head=head_branch.commit.sha,
title=title,
body=body,
maintainer_can_modify=maintainer_can_modify,
draft=draft,
issue=issue,
)
return pr
@staticmethod
async def create_issue(
repo_name: str,
title: str,
body: Optional[str] = None,
assignee: NamedUser | Optional[str] = None,
milestone: Optional[Milestone] = None,
labels: list[Label] | Optional[list[str]] = None,
assignees: Optional[list[str]] | list[NamedUser] = None,
access_token: Optional[str] = None,
auth: Optional[Auth] = None,
) -> Issue:
"""
Creates an issue in the specified repository.
Args:
repo_name (str): The full repository name (user/repo) where the issue will be created.
title (str): The title of the issue.
body (Optional[str], optional): The body of the issue. Defaults to None.
assignee (Union[NamedUser, str], optional): The assignee for the issue, either as a NamedUser object or their username. Defaults to None.
milestone (Optional[Milestone], optional): The milestone to associate with the issue. Defaults to None.
labels (Union[list[Label], list[str]], optional): The labels to associate with the issue, either as Label objects or their names. Defaults to None.
assignees (Union[list[str], list[NamedUser]], optional): The list of usernames or NamedUser objects to assign to the issue. Defaults to None.
access_token (Optional[str], optional): The access token for authentication. Defaults to None. Visit `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`, `https://github.com/PyGithub/PyGithub/blob/main/doc/examples/Authentication.rst`.
auth (Optional[Auth], optional): The authentication method. Defaults to None. Visit `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`
Returns:
Issue: The created issue object.
"""
body = body or NotSet
assignee = assignee or NotSet
milestone = milestone or NotSet
labels = labels or NotSet
assignees = assignees or NotSet
if not auth and not access_token:
raise ValueError('`access_token` is invalid. Visit: "https://github.com/settings/tokens"')
auth = auth or Auth.Token(access_token)
g = Github(auth=auth)
repo = g.get_repo(repo_name)
x_ratelimit_remaining = repo.raw_headers.get("x-ratelimit-remaining")
if (
x_ratelimit_remaining
and bool(re.match(r"^-?\d+$", x_ratelimit_remaining))
and int(x_ratelimit_remaining) <= 0
):
raise RateLimitError()
issue = repo.create_issue(
title=title,
body=body,
assignee=assignee,
milestone=milestone,
labels=labels,
assignees=assignees,
)
return issue
@staticmethod
async def get_repos(access_token: Optional[str] = None, auth: Optional[Auth] = None) -> List[str]:
"""
Fetches a list of public repositories belonging to the authenticated user.
Args:
access_token (Optional[str], optional): The access token for authentication. Defaults to None.
Visit `https://github.com/settings/tokens` for obtaining a personal access token.
auth (Optional[Auth], optional): The authentication method. Defaults to None.
Visit `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html` for more information.
Returns:
List[str]: A list of full names of the public repositories belonging to the user.
"""
auth = auth or Auth.Token(access_token)
git = Github(auth=auth)
user = git.get_user()
v = user.get_repos(visibility="public")
return [i.full_name for i in v]

71
tests/metagpt/tools/libs/test_git.py
View file

@ -1,7 +1,11 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import annotations
import os
import pytest
from github import Auth, Github
from pydantic import BaseModel
from metagpt.tools.libs.git import git_checkout, git_clone
@ -13,10 +17,15 @@ class SWEBenchItem(BaseModel):
repo: str
def get_env(key):
return os.environ.get(key)
@pytest.mark.asyncio
@pytest.mark.parametrize(
["url", "commit_id"], [("https://github.com/sqlfluff/sqlfluff.git", "d19de0ecd16d298f9e3bfb91da122734c40c01e5")]
)
@pytest.mark.skip
async def test_git(url: str, commit_id: str):
repo_dir = await git_clone(url)
assert repo_dir
@ -27,5 +36,67 @@ async def test_git(url: str, commit_id: str):
repo.delete_repository()
@pytest.mark.skip
def test_login():
auth = Auth.Login(get_env("GITHUB_USER"), get_env("GITHUB_PWD"))
g = Github(auth=auth)
repo = g.get_repo("geekan/MetaGPT")
topics = repo.get_topics()
assert topics
open_issues = repo.get_issues(state="open")
issues = [i for i in open_issues]
assert issues
@pytest.mark.skip
@pytest.mark.asyncio
async def test_new_issue():
issue = await GitRepository.create_issue(
repo_name="iorisa/MetaGPT",
title="This is a new issue",
body="This is the issue body",
access_token=get_env("GITHUB_PERSONAL_ACCESS_TOKEN"),
)
print(issue)
assert issue.number
pass
@pytest.mark.skip
@pytest.mark.asyncio
async def test_new_pr():
body = """
>>> SUMMARY
>>> Change HTTP library used to send requests
>>>
>>> TESTS
>>> - [x] Send 'GET' request
>>> - [x] Send 'POST' request with/without body
"""
pr = await GitRepository.create_pull(
repo_name="iorisa/MetaGPT",
base="send18",
head="fixbug/gbk",
title="Test pr",
body=body,
access_token=get_env("GITHUB_PERSONAL_ACCESS_TOKEN"),
)
print(pr)
assert pr
@pytest.mark.skip
def test_auth():
access_token = get_env("GITHUB_PERSONAL_ACCESS_TOKEN")
auth = Auth.Token(access_token)
g = Github(auth=auth)
u = g.get_user()
v = u.get_repos(visibility="public")
a = [i.full_name for i in v]
assert a
print(a)
pass
if __name__ == "__main__":
pytest.main([__file__, "-s"])