Implement 'git-obs repo list' command

osc/gitea_api/connection.py

@@ -1,8 +1,11 @@
 import copy
 import http.client
 import json
+import re
 import time
 import urllib.parse
+from typing import Dict
+from typing import Generator
 from typing import Optional
 
 import urllib3
@@ -12,6 +15,19 @@ import urllib3.response
 from .conf import Login
 
 
+RE_HTTP_HEADER_LINK = re.compile('<(?P<url>.*?)>; rel="(?P<rel>.*?)",?')
+
+
+def parse_http_header_link(link: str) -> Dict[str, str]:
+    """
+    Parse RFC8288 "link" http headers into {"rel": "url"}
+    """
+    result = {}
+    for match in RE_HTTP_HEADER_LINK.findall(link):
+        result[match[1]] = match[0]
+    return result
+
+
 class GiteaHTTPResponse:
     """
     A ``urllib3.response.HTTPResponse`` wrapper
@@ -153,3 +169,24 @@ class Connection:
             raise response_to_exception(response, context=context)
 
         return response
+
+    def request_all_pages(
+        self, method, url, json_data: Optional[dict] = None, *, context: Optional[dict] = None
+    ) -> Generator[GiteaHTTPResponse, None, None]:
+        """
+        Make a request and yield ``GiteaHTTPResponse`` instances for each page.
+        Arguments are forwarded to the underlying ``request()`` call.
+        """
+
+        while True:
+            response = self.request(method, url, json_data=json_data, context=context)
+            yield response
+
+            if "link" not in response.headers:
+                break
+
+            links = parse_http_header_link(response.headers["link"])
+            if "next" in links:
+                url = links["next"]
+            else:
+                break

osc/gitea_api/repo.py

@@ -1,6 +1,8 @@
+import functools
 import os
 import re
 import subprocess
+from typing import List
 from typing import Optional
 from typing import Tuple
 
@@ -9,11 +11,18 @@ from .connection import GiteaHTTPResponse
 from .user import User
 
 
+@functools.total_ordering
 class Repo:
     def __init__(self, data: dict, *, response: Optional[GiteaHTTPResponse] = None):
         self._data = data
         self._response = response
 
+    def __eq__(self, other):
+        (self.owner, self.repo) == (other.owner, other.repo)
+
+    def __lt__(self, other):
+        (self.owner, self.repo) < (other.owner, other.repo)
+
     @property
     def owner(self) -> str:
         return self._data["owner"]["login"]
@@ -179,3 +188,37 @@ class Repo:
             subprocess.run(cmd, cwd=cwd, check=True)
 
         return directory_abspath
+
+    @classmethod
+    def list_org_repos(cls, conn: Connection, owner: str) -> List["Repo"]:
+        """
+        List repos owned by an organization.
+
+        :param conn: Gitea ``Connection`` instance.
+        """
+        q = {
+            # XXX: limit works in range 1..50, setting it any higher doesn't help, we need to handle paginated results
+            "limit": 10**6,
+        }
+        url = conn.makeurl("orgs", owner, "repos", query=q)
+        obj_list = []
+        for response in conn.request_all_pages("GET", url):
+            obj_list.extend([cls(i, response=response) for i in response.json()])
+        return obj_list
+
+    @classmethod
+    def list_user_repos(cls, conn: Connection, owner: str) -> List["Repo"]:
+        """
+        List repos owned by a user.
+
+        :param conn: Gitea ``Connection`` instance.
+        """
+        q = {
+            # XXX: limit works in range 1..50, setting it any higher doesn't help, we need to handle paginated results
+            "limit": 10**6,
+        }
+        url = conn.makeurl("users", owner, "repos", query=q)
+        obj_list = []
+        for response in conn.request_all_pages("GET", url):
+            obj_list.extend([cls(i, response=response) for i in response.json()])
+        return obj_list