diff --git a/interface/forges/gitea.py b/interface/forges/gitea.py index b9dfe3c..6f1198a 100644 --- a/interface/forges/gitea.py +++ b/interface/forges/gitea.py @@ -27,35 +27,36 @@ from forge import CreateIssue, Forge, RepositoryInfo, Comment from forge import Notification, NotificationResp, CreatePullrequest -ISSSUE="Issue" -PULL ="pull" +ISSUE = "Issue" +PULL = "pull" COMMIT = "commit" REPOSITORY = "repository" + class Gitea(Forge): def __init__(self): self.host = urlparse(utils.clean_url(local_settings.GITEA_HOST)) def _auth(self): - return {'Authorization': format("token %s" % (local_settings.GITEA_API_KEY))} + return {"Authorization": format("token %s" % (local_settings.GITEA_API_KEY))} def _get_url(self, path: str) -> str: prefix = "/api/v1/" - if path.startswith('/'): - path=path[1:] + if path.startswith("/"): + path = path[1:] path = format("%s%s" % (prefix, path)) url = urlunparse((self.host.scheme, self.host.netloc, path, "", "", "")) return url def get_issues(self, owner: str, repo: str, *args, **kwargs): - """ Get issues on a repository. Supports pagination via 'page' optional param""" + """Get issues on a repository. Supports pagination via 'page' optional param""" query = {} - since = kwargs.get('since') + since = kwargs.get("since") if since is not None: query["since"] = since - page = kwargs.get('page') + page = kwargs.get("page") if page is not None: query["page"] = page @@ -67,7 +68,7 @@ def get_issues(self, owner: str, repo: str, *args, **kwargs): return response.json() def create_issue(self, owner: str, repo: str, issue: CreateIssue): - """ Creates issue on a repository""" + """Creates issue on a repository""" url = self._get_url(format("/repos/%s/%s/issues" % (owner, repo))) headers = self._auth() @@ -83,16 +84,16 @@ def _into_repository(self, data) -> RepositoryInfo: return info def get_repository(self, owner: str, repo: str) -> RepositoryInfo: - """ Get repository details""" + """Get repository details""" url = self._get_url(format("/repos/%s/%s" % (owner, repo))) response = requests.request("GET", url) - data = response.json(); + data = response.json() info = self._into_repository(data) return info.get_payload() def create_repository(self, repo: str, description: str): url = self._get_url("/user/repos/") - payload = { "name" : repo, "description": description} + payload = {"name": repo, "description": description} headers = self._auth() _response = requests.request("POST", url, json=payload, headers=headers) @@ -101,10 +102,9 @@ def subscribe(self, owner: str, repo: str): headers = self._auth() _response = requests.request("PUT", url, headers=headers) - def get_notifications(self, since: datetime.datetime) -> NotificationResp: query = {} - query["since"] = rfc3339(since) + query["since"] = rfc3339(since) url = self._get_url("/notifications") headers = self._auth() response = requests.request("GET", url, params=query, headers=headers) @@ -126,7 +126,7 @@ def get_notifications(self, since: datetime.datetime) -> NotificationResp: if notification_type == REPOSITORY: print(n) - if notification_type == ISSSUE: + if notification_type == ISSUE: comment_url = subject["latest_comment_url"] print(comment_url) if len(comment_url) != 0: @@ -151,7 +151,7 @@ def create_pull_request(self, owner: str, repo: str, pr: CreatePullrequest): url = self._get_url(format("/repos/%s/%s/pulls" % (owner, repo))) headers = self._auth() - payload = pr.get_payload() + payload = pr.get_payload() for key in ["repo", "owner"]: del payload[key] @@ -162,16 +162,16 @@ def create_pull_request(self, owner: str, repo: str, pr: CreatePullrequest): response = requests.request("POST", url, json=payload, headers=headers) return response.json()["html_url"] - def fork(self, owner: str, repo:str): - """ Fork a repository """ + def fork(self, owner: str, repo: str): + """Fork a repository""" url = self._get_url(format("/repos/%s/%s/forks" % (owner, repo))) print(url) headers = self._auth() - payload = {"oarganization" :"bot"} + payload = {"oarganization": "bot"} _response = requests.request("POST", url, json=payload, headers=headers) -#if __name__ == "__main__": +# if __name__ == "__main__": # owner = "realaravinth" # repo = "tmp" # g = Gitea() diff --git a/interface/forges/github.py b/interface/forges/github.py index cd749df..ac70bd4 100644 --- a/interface/forges/github.py +++ b/interface/forges/github.py @@ -14,21 +14,32 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +from dateutil.parser import parse as date_parse +import datetime from urllib.parse import urlparse, urlunparse import requests import sys +from rfc3339 import rfc3339 + sys.path.append("..") import local_settings import utils -from forge import Forge, CreateIssue, RepositoryInfo +from forge import CreateIssue, Forge, RepositoryInfo, Comment +from forge import Notification, NotificationResp, CreatePullrequest +ISSUE = "Issue" +PULL = "pull" +COMMIT = "commit" +REPOSITORY = "repository" class GitHub(Forge): def __init__(self): + """Initializes the class variables""" self.host = urlparse(utils.clean_url(local_settings.GITHUB_HOST)) def _get_url(self, path: str) -> str: + """Retrieves the forge url""" if path.startswith("/"): path = path[1:] @@ -36,13 +47,15 @@ def _get_url(self, path: str) -> str: return url def _auth(self): + """Authorizes the request with a token""" return {"Authorization": format("token %s" % (local_settings.GITHUB_API_KEY))} def get_issues(self, owner: str, repo: str, *args, **kwargs): """Get the issues present in a provided repository""" # Defining a formatted url with the repo details - url = format("https://api.github.com/repos/%s/%s/issues" % (owner, repo)) + url = self._get_url(format("/repos/%s/%s/issues" % (owner, repo))) + print(url) # Requesting the issues present in the repo # GitHub provides a paginated response for 30 @@ -85,16 +98,103 @@ def get_repository(self, owner: str, repo: str) -> RepositoryInfo: print("Payload deets", info.get_payload()) return info + def create_repository(self, repo: str, description: str): + """Creates a repository in the users forge""" + url = self._get_url("/users/repos/") + payload = {"name": repo, "description": description} + headers = self._auth() + _response = requests.request("POST", url, json=payload, headers=headers) + + def subscribe(self, owner: str, repo: str): + """Subscribes/watches a repository""" + url = self._get_url(format("/repos/%s/%s/subscription" % (owner, repo))) + headers = self._auth() + _response = requests.request("PUT", url, headers=headers) + + def get_notifications(self, since: datetime.datetime) -> NotificationResp: + """Notifications for watched repositories""" + query = {} + query["since"] = rfc3339(since) + url = self._get_url("/notifications") + headers = self._auth() + response = requests.request("GET", url, params=query, headers=headers) + notifications = response.json() + last_read = "" + val = [] + for n in notifications: + rn = Notification() + subject = n["subject"] + notification_type = subject["type"] + + last_read = n["updated_at"] + rn.set_updated_at(last_read) + rn.set_type(notification_type) + rn.set_title(subject["title"]) + rn.set_state(subject["state"]) + rn.set_repo_url(n["repository"]["html_url"]) + + if notification_type == REPOSITORY: + print(n) + if notification_type == ISSUE: + comment_url = subject["latest_comment_url"] + print(comment_url) + if len(comment_url) != 0: + resp = requests.request("GET", comment_url) + comment = resp.json() + if date_parse(comment["updated_at"]) > since: + c = Comment() + c.set_updated_at(comment["updated_at"]) + c.set_author(comment["user"]["login"]) + c.set_body(comment["body"]) + pr_url = comment["pull_request_url"] + if len(comment["pull_request_url"]) == 0: + c.set_url(comment["issue_url"]) + else: + url = pr_url + c.set_url(comment["pull_request_url"]) + rn.set_comment(c) + val.append(rn) + print(last_read) + return NotificationResp(val, date_parse(last_read)) + + def create_pull_request(self, owner: str, repo: str, pr: CreatePullrequest): + """Creates a POST request for the Pull Request""" + url = self._get_url(format("/repos/%s/%s/pulls" % (owner, repo))) + headers = self._auth() + + payload = pr.get_payload() + for key in ["repo", "owner"]: + del payload[key] + + payload["assignees"] = [] + payload["lables"] = [0] + payload["milestones"] = 0 + + response = requests.request("POST", url, json=payload, headers=headers) + print(response.json()) + return response.json() + if __name__ == "__main__": # Testing the API primitively with a simple call + + # Setting owner and repo owner = "dat-adi" repo = "tmp" g = GitHub() - issue = CreateIssue() - issue.set_title("testing yet again") - print(g.create_issue(owner, repo, issue)) + print("HOST : ", g.host) + print("GET URL : ", g._get_url(f"/repos/{owner}/{repo}")) + print("AUTH : ", g._auth()) +# print("ISSUES : ", g.get_issues(owner, repo)) +# print("GET REPO : ", g.get_repository("dat-adi", "tmp")) + +# issue = CreateIssue() +# issue.set_title("another test, to be extra sure.") +# print(g.create_issue(owner, repo, issue)) +# print("SUBSCRIBE : ", g.subscribe("dat-adi", "tmp")) + print( + "INTO REPOSITORY : ", g._into_repository( { "name": "G V Datta Adithya", @@ -103,3 +203,18 @@ def get_repository(self, owner: str, repo: str) -> RepositoryInfo: } ) ) + + + """ + notifications = g.get_notifications(since=date_parse("2021-10-10T17:06:02+05:30")) + print(notifications) + + pr = CreatePullrequest() + pr.set_owner(owner) + pr.set_repo(repo) + pr.set_base("main") + pr.set_head("dev") + pr.set_title("How many more tests?") + pr.set_body("Does this create a PR?") + print(g.create_pull_request(owner, repo, pr)) + """