Notifications and PRs

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()

interface/forges/github.py

@@ -14,35 +14,48 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+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:]
 
         url = urlunparse((self.host.scheme, self.host.netloc, path, "", "", ""))
         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))
+    """