CENG-452: Add --sort flag for package list command (#185)

Author: BartoszBlizniak

cloudsmith_cli/cli/commands/list_.py

@@ -127,8 +127,34 @@ def entitlements_(*args, **kwargs):  # pylint: disable=missing-docstring
     "--query",
     help=("A boolean-like search term for querying package attributes."),
 )
+@click.option(
+    "--sort",
+    type=click.Choice(
+        [
+            "date",
+            "-date",
+            "downloads",
+            "-downloads",
+            "format",
+            "-format",
+            "name",
+            "-name",
+            "scan",
+            "-scan",
+            "scan_date",
+            "-scan_date",
+            "size",
+            "-size",
+            "status",
+            "-status",
+            "version",
+            "-version",
+        ]
+    ),
+    help=("Sort packages by field. Prefix with '-' for descending order."),
+)
 @click.pass_context
-def packages(ctx, opts, owner_repo, page, page_size, query):
+def packages(ctx, opts, owner_repo, page, page_size, query, sort):
     """
     List packages for a repository.
 
@@ -165,6 +191,18 @@ def packages(ctx, opts, owner_repo, page, page_size, query):
 
     --query 'name:^foo$ filename:.zip$ architecture:~x86'
 
+    You can sort the results using --sort with these fields:
+      - date/-date: Sort by creation date
+      - downloads/-downloads: Sort by download count
+      - format/-format: Sort by package format
+      - name/-name: Sort by package name
+      - scan/-scan: Sort by scan status
+      - scan_date/-scan_date: Sort by last scan date
+      - size/-size: Sort by package size
+      - status/-status: Sort by package status
+      - version/-version: Sort by version
+
+    Prefix any field with '-' for descending order (e.g. -date for newest first).
     """
     owner, repo = owner_repo
 
@@ -177,7 +215,12 @@ def packages(ctx, opts, owner_repo, page, page_size, query):
     with handle_api_exceptions(ctx, opts=opts, context_msg=context_msg):
         with maybe_spinner(opts):
             packages_, page_info = list_packages(
-                owner=owner, repo=repo, page=page, page_size=page_size, query=query
+                owner=owner,
+                repo=repo,
+                page=page,
+                page_size=page_size,
+                query=query,
+                sort=sort,
             )
 
     click.secho("OK", fg="green", err=use_stderr)

cloudsmith_cli/cli/tests/commands/test_package_commands.py

@@ -78,3 +78,88 @@ def test_push_and_delete_raw_package(
     )
     data = json.loads(result.output)["data"]
     assert len(data) == 0
+
+
+@pytest.mark.usefixtures("set_api_key_env_var", "set_api_host_env_var")
+def test_list_packages_with_sort(runner, organization, tmp_repository, tmp_path):
+    """Test listing packages with different sort options."""
+    org_repo = f'{organization}/{tmp_repository["slug"]}'
+
+    # Create and push two packages with different names
+    for name in ["aaa", "zzz"]:
+        pkg_file = tmp_path / f"{name}.txt"
+        with open(pkg_file, "wb") as f:
+            f.write(b"test content")
+
+        runner.invoke(
+            push,
+            args=["raw", org_repo, str(pkg_file.resolve())],
+            catch_exceptions=False,
+        )
+
+        # Wait for package to sync
+        result = runner.invoke(
+            list_, args=["pkgs", org_repo, "-F", "json"], catch_exceptions=False
+        )
+        data = json.loads(result.output)["data"]
+        pkg_slug = next(pkg["slug"] for pkg in data if pkg["filename"] == pkg_file.name)
+        org_repo_package = f"{org_repo}/{pkg_slug}"
+
+        for _ in range(10):
+            time.sleep(5)
+            result = runner.invoke(
+                status, args=[org_repo_package], catch_exceptions=False
+            )
+            if "Fully Synchronised" in result.output:
+                break
+        else:
+            raise TimeoutError("Test timed out waiting for package sync")
+
+    # Test ascending sort by name
+    result = runner.invoke(
+        list_,
+        args=["pkgs", org_repo, "--sort", "name", "-F", "json"],
+        catch_exceptions=False,
+    )
+    data = json.loads(result.output)["data"]
+    assert len(data) == 2
+    assert data[0]["filename"] == "aaa.txt"
+    assert data[1]["filename"] == "zzz.txt"
+
+    # Test descending sort by name
+    result = runner.invoke(
+        list_,
+        args=["pkgs", org_repo, "--sort", "-name", "-F", "json"],
+        catch_exceptions=False,
+    )
+    data = json.loads(result.output)["data"]
+    assert len(data) == 2
+    assert data[0]["filename"] == "zzz.txt"
+    assert data[1]["filename"] == "aaa.txt"
+
+    # Test sort by date (newest first)
+    result = runner.invoke(
+        list_,
+        args=["pkgs", org_repo, "--sort", "-date", "-F", "json"],
+        catch_exceptions=False,
+    )
+    data = json.loads(result.output)["data"]
+    assert len(data) == 2
+    assert data[0]["filename"] == "zzz.txt"  # Last uploaded
+    assert data[1]["filename"] == "aaa.txt"  # First uploaded
+
+    # Cleanup - delete both packages
+    for pkg in data:
+        org_repo_package = f"{org_repo}/{pkg['slug']}"
+        runner.invoke(delete, args=["-y", org_repo_package], catch_exceptions=False)
+
+        # Wait for deletion
+        for _ in range(10):
+            time.sleep(5)
+            result = runner.invoke(
+                status, args=[org_repo_package], catch_exceptions=False
+            )
+            if "status: 404 - Not Found" in result.output:
+                break
+        else:
+            raise TimeoutError("Test timed out waiting for package deletion")

cloudsmith_cli/core/api/packages.py

@@ -222,6 +222,7 @@ def list_packages(owner, repo, **kwargs):
     api_kwargs = {}
     api_kwargs.update(utils.get_page_kwargs(**kwargs))
     api_kwargs.update(utils.get_query_kwargs(**kwargs))
+    api_kwargs.update(utils.get_sort_kwargs(**kwargs))
 
     with catch_raise_api_exception():
         data, _, headers = client.packages_list_with_http_info(

cloudsmith_cli/core/utils.py

@@ -82,3 +82,14 @@ def get_query_kwargs(**kwargs):
         query_kwargs["query"] = query
 
     return query_kwargs
+
+
+def get_sort_kwargs(**kwargs):
+    """Construct sort kwargs (if present)."""
+    sort_kwargs = {}
+
+    sort = kwargs.get("sort")
+    if sort:
+        sort_kwargs["sort"] = sort
+
+    return sort_kwargs