Add --limit to queue flush

projects/fal/src/fal/cli/queue.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import argparse
 import json
 from http import HTTPStatus
 
@@ -8,6 +9,13 @@
 from .parser import FalClientParser, get_output_parser
 
 
+def _positive_int(value: str) -> int:
+    parsed = int(value)
+    if parsed < 1:
+        raise argparse.ArgumentTypeError("must be a positive integer")
+    return parsed
+
+
 def _queue_size(args):
     from fal.api.client import SyncServerlessClient
     from fal.api.deploy import _get_user
@@ -63,15 +71,17 @@ def _queue_flush(args):
 
     client = SyncServerlessClient(host=args.host, team=args.team)._create_rest_client()
     user = _get_user(client)
-    caller_user_id = args.caller_user_id
+    params = {}
+    if args.caller_user_id:
+        params["caller_user_id"] = args.caller_user_id
+    if args.limit is not None:
+        params["limit"] = args.limit
 
     url = f"{client.base_url}/applications/{user.username}/{args.app_name}/queue"
     headers = client.get_headers()
 
     with httpx.Client(base_url=client.base_url, headers=headers, timeout=300) as c:
-        resp = c.delete(
-            url, params={"caller_user_id": caller_user_id} if caller_user_id else None
-        )
+        resp = c.delete(url, params=params or None)
 
     if resp.status_code != HTTPStatus.OK:
         try:
@@ -118,7 +128,7 @@ def add_parser(main_subparsers, parents):
     )
     size_parser.set_defaults(func=_queue_size)
 
-    flush_help = "Flush all pending requests in an application queue."
+    flush_help = "Flush pending requests in an application queue."
     flush_parser = subparsers.add_parser(
         "flush",
         description=flush_help,
@@ -129,13 +139,17 @@ def add_parser(main_subparsers, parents):
         "app_name",
         help="Application name.",
     )
-    flush_parser.add_argument(
+    flush_filter = flush_parser.add_mutually_exclusive_group()
+    flush_filter.add_argument(
         "--caller-user-id",
         default=None,
         type=str,
-        help=(
-            "Only flush requests from this user ID. "
-            "If not provided, all requests will be flushed."
-        ),
+        help="Only flush requests from this user ID.",
+    )
+    flush_filter.add_argument(
+        "--limit",
+        default=None,
+        type=_positive_int,
+        help="Flush only the first N pending requests.",
     )
     flush_parser.set_defaults(func=_queue_flush)

projects/fal/tests/unit/cli/test_queue.py

@@ -0,0 +1,66 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from fal.cli.main import parse_args
+from fal.cli.parser import FalParserExit
+from fal.cli.queue import _queue_flush
+
+
+def test_queue_flush_with_limit():
+    args = parse_args(["queue", "flush", "my-app", "--limit", "2"])
+
+    assert args.func == _queue_flush
+    assert args.app_name == "my-app"
+    assert args.limit == 2
+    assert args.caller_user_id is None
+
+
+def test_queue_flush_sends_limit():
+    args = parse_args(["queue", "flush", "my-app", "--limit", "2"])
+    args.console = MagicMock()
+    rest_client = SimpleNamespace(
+        base_url="https://api.example.com",
+        get_headers=lambda: {"Authorization": "Key test"},
+    )
+    response = SimpleNamespace(status_code=200)
+    http_client = MagicMock()
+    http_client.__enter__.return_value.delete.return_value = response
+
+    with (
+        patch("fal.api.client.SyncServerlessClient") as client_type,
+        patch(
+            "fal.api.deploy._get_user",
+            return_value=SimpleNamespace(username="owner"),
+        ),
+        patch("fal.cli.queue.httpx.Client", return_value=http_client),
+    ):
+        client_type.return_value._create_rest_client.return_value = rest_client
+        args.func(args)
+
+    http_client.__enter__.return_value.delete.assert_called_once_with(
+        "https://api.example.com/applications/owner/my-app/queue",
+        params={"limit": 2},
+    )
+
+
+@pytest.mark.parametrize("limit", ["0", "-1"])
+def test_queue_flush_rejects_non_positive_limit(limit: str):
+    with pytest.raises(FalParserExit):
+        parse_args(["queue", "flush", "my-app", "--limit", limit])
+
+
+def test_queue_flush_rejects_limit_with_caller_user_id():
+    with pytest.raises(FalParserExit):
+        parse_args(
+            [
+                "queue",
+                "flush",
+                "my-app",
+                "--limit",
+                "2",
+                "--caller-user-id",
+                "caller",
+            ]
+        )