Make use of paginated API by default on list-projects and list-jobs

Author: suricactus

qfieldcloud_sdk/cli.py

@@ -1,7 +1,7 @@
 import collections
 import platform
 from enum import Enum
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 import click
 
@@ -33,6 +33,26 @@ def list_commands(self, ctx):
         return self.commands
 
 
+def paginated(command):
+    command = click.option(
+        "-o",
+        "--offset",
+        type=int,
+        default=None,
+        is_flag=False,
+        help="Offsets the given number of records in the paginated JSON response.",
+    )(command)
+    command = click.option(
+        "-l",
+        "--limit",
+        type=int,
+        default=None,
+        is_flag=False,
+        help="Limits the number of records to return in the paginated JSON response.",
+    )(command)
+    return command
+
+
 @click.group(cls=OrderedGroup)
 @click.option(
     "-U",
@@ -149,35 +169,23 @@ def logout(ctx):
 
 
 @cli.command()
-@click.option(
-    "-o",
-    "--offset",
-    type=int,
-    default=None,
-    is_flag=False,
-    help="Offsets the given number of projects in the paginated JSON response",
-)
-@click.option(
-    "-l",
-    "--limit",
-    type=int,
-    default=None,
-    is_flag=False,
-    help="Limits the number of projects to return in the paginated JSON response",
-)
+@paginated
 @click.option(
     "--include-public/--no-public",
     default=False,
     is_flag=True,
     help="Includes the public project in the list. Default: False",
 )
 @click.pass_context
-def list_projects(ctx, **opts):
+def list_projects(ctx, include_public, **opts):
     """List QFieldCloud projects."""
 
     log("Listing projects…")
 
-    projects: List[Dict[str, Any]] = ctx.obj["client"].list_projects(**opts)
+    projects: List[Dict[str, Any]] = ctx.obj["client"].list_projects(
+        include_public,
+        sdk.Pagination(**opts),
+    )
 
     if ctx.obj["format_json"]:
         print_json(projects)
@@ -381,29 +389,18 @@ def delete_files(ctx, project_id, paths, throw_on_error):
     type=sdk.JobTypes,
     help="Job type. One of package, delta_apply or process_projectfile.",
 )
-@click.option(
-    "-o",
-    "--offset",
-    type=int,
-    default=None,
-    is_flag=False,
-    help="Offsets the given number of projects in the paginated JSON response",
-)
-@click.option(
-    "-l",
-    "--limit",
-    type=int,
-    default=None,
-    is_flag=False,
-    help="Limits the number of projects to return in the paginated JSON response",
-)
+@paginated
 @click.pass_context
-def list_jobs(ctx, project_id, **opts):
+def list_jobs(ctx, project_id, job_type: Optional[sdk.JobTypes], **opts):
     """List project jobs."""
 
     log(f'Listing project "{project_id}" jobs…')
 
-    jobs: List[Dict[Any]] = ctx.obj["client"].list_jobs(project_id, **opts)
+    jobs: List[Dict[Any]] = ctx.obj["client"].list_jobs(
+        project_id,
+        job_type,
+        sdk.Pagination(**opts),
+    )
 
     if ctx.obj["format_json"]:
         print_json(jobs)

qfieldcloud_sdk/interfaces.py

@@ -15,14 +15,28 @@ def __getitem__(self, k: str) -> Any:
 class QfcMockResponse(requests.Response):
     def __init__(self, **kwargs):
         self.request_kwargs = kwargs
+        self.url = kwargs["url"]
         self.limit = kwargs.get("limit", 5)
         self.total = self.limit * 2
         self.headers = {
             "X-Total-Count": self.total,
-            "X-Next-Page": "next_url",
-            "X-Previous-Page": "previous_url",
         }
 
+        limit = kwargs["params"].get("limit")
+        offset = kwargs["params"].get("offset", 0)
+        prev_url = None
+        next_url = None
+        if limit:
+            if offset == 0:
+                prev_url = None
+                next_url = f"{self.url}?limit={limit}&offset={limit}"
+            else:
+                prev_url = f"{self.url}?limit={limit}&offset=0"
+                next_url = None
+
+        self.headers["X-Previous-Page"] = prev_url
+        self.headers["X-Next-Page"] = next_url
+
     def json(self) -> Union[QfcMockItem, List[QfcMockItem]]:
         if self.request_kwargs["method"] == "GET":
             return [QfcMockItem(id=n) for n in range(self.total)]