REFACTOR: Promote backend service & CLI client defaults to ClassVar

pyrit/backend/services/attack_service.py

@@ -21,7 +21,7 @@
 from datetime import datetime, timezone
 from functools import lru_cache
 from pathlib import Path
-from typing import Any, Literal, cast
+from typing import Any, ClassVar, Literal, cast
 from urllib.parse import parse_qs, urlparse
 
 from pyrit.backend.mappers import (
@@ -71,6 +71,8 @@ class AttackService:
     Uses PyRIT memory (database) as the source of truth via AttackResult.
     """
 
+    DEFAULT_LIST_LIMIT: ClassVar[int] = 20
+
     def __init__(self) -> None:
         """Initialize the attack service."""
         self._memory = CentralMemory.get_memory_instance()
@@ -90,7 +92,7 @@ async def list_attacks_async(
         labels: dict[str, str | Sequence[str]] | None = None,
         min_turns: int | None = None,
         max_turns: int | None = None,
-        limit: int = 20,
+        limit: int | None = None,
         cursor: str | None = None,
     ) -> AttackListResponse:
         """
@@ -119,12 +121,16 @@ async def list_attacks_async(
                 each name).
             min_turns: Filter by minimum executed turns.
             max_turns: Filter by maximum executed turns.
-            limit: Maximum items to return.
+            limit: Maximum items to return. Defaults to ``DEFAULT_LIST_LIMIT`` (20)
+                when ``None``.
             cursor: Pagination cursor.
 
         Returns:
             AttackListResponse with filtered and paginated attack summaries.
         """
+        if limit is None:
+            limit = self.DEFAULT_LIST_LIMIT
+
         # Phase 1: Query + lightweight filtering (no pieces needed)
         # Coerce an empty converter_types list to None so it behaves as "no filter" at
         # this layer — the "attacks with no converters" case is expressed through

pyrit/backend/services/initializer_service.py

@@ -10,6 +10,7 @@
 
 import logging
 from functools import lru_cache
+from typing import ClassVar
 
 from pyrit.backend.models.common import PaginationInfo
 from pyrit.backend.models.initializers import (
@@ -55,26 +56,32 @@ class InitializerService:
     Uses InitializerRegistry as the source of truth for initializer metadata.
     """
 
+    DEFAULT_LIST_LIMIT: ClassVar[int] = 50
+
     def __init__(self) -> None:
         """Initialize the initializer service."""
         self._registry = InitializerRegistry.get_registry_singleton()
 
     async def list_initializers_async(
         self,
         *,
-        limit: int = 50,
+        limit: int | None = None,
         cursor: str | None = None,
     ) -> ListRegisteredInitializersResponse:
         """
         List all available initializers with pagination.
 
         Args:
-            limit: Maximum items to return per page.
+            limit: Maximum items to return per page. Defaults to
+                ``DEFAULT_LIST_LIMIT`` (50) when ``None``.
             cursor: Pagination cursor (initializer_name to start after).
 
         Returns:
             ListRegisteredInitializersResponse with paginated initializer summaries.
         """
+        if limit is None:
+            limit = self.DEFAULT_LIST_LIMIT
+
         all_metadata = self._registry.list_metadata()
         all_summaries = [_metadata_to_registered_initializer(m) for m in all_metadata]
 

pyrit/backend/services/scenario_run_service.py

@@ -12,7 +12,7 @@
 import contextlib
 import logging
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from pyrit.backend.models.scenarios import (
     RunScenarioRequest,
@@ -31,8 +31,6 @@
 
 logger = logging.getLogger(__name__)
 
-_DEFAULT_MAX_CONCURRENT_RUNS = 3
-
 
 @dataclass
 class _ActiveTask:
@@ -52,8 +50,20 @@ class ScenarioRunService:
     Keeps an in-memory dict only for active asyncio tasks (cancellation support).
     """
 
-    def __init__(self, *, max_concurrent_runs: int = _DEFAULT_MAX_CONCURRENT_RUNS) -> None:
-        """Initialize the scenario run service."""
+    DEFAULT_MAX_CONCURRENT_RUNS: ClassVar[int] = 3
+    DEFAULT_LIST_LIMIT: ClassVar[int] = 100
+
+    def __init__(self, *, max_concurrent_runs: int | None = None) -> None:
+        """
+        Initialize the scenario run service.
+
+        Args:
+            max_concurrent_runs: Maximum number of scenario runs allowed
+                concurrently. Defaults to ``DEFAULT_MAX_CONCURRENT_RUNS`` (3)
+                when ``None``.
+        """
+        if max_concurrent_runs is None:
+            max_concurrent_runs = self.DEFAULT_MAX_CONCURRENT_RUNS
         self._max_concurrent_runs = max_concurrent_runs
         self._memory = CentralMemory.get_memory_instance()
         self._active_tasks: dict[str, _ActiveTask] = {}
@@ -131,16 +141,20 @@ def get_run(self, *, scenario_result_id: str) -> ScenarioRunSummary | None:
         """
         return self._build_response(scenario_result_id=scenario_result_id)
 
-    def list_runs(self, *, limit: int = 100) -> ScenarioRunListResponse:
+    def list_runs(self, *, limit: int | None = None) -> ScenarioRunListResponse:
         """
         List scenario runs by querying the database (most recent first).
 
         Args:
-            limit (int): Maximum number of runs to return. Defaults to 100.
+            limit (int | None): Maximum number of runs to return. Defaults to
+                ``DEFAULT_LIST_LIMIT`` (100) when ``None``.
 
         Returns:
             ScenarioRunListResponse with runs.
         """
+        if limit is None:
+            limit = self.DEFAULT_LIST_LIMIT
+
         # This is expensive, and we don't need all the data. At some point
         # we may want to add a lightweight "list" query to the DB layer that only
         results = self._memory.get_scenario_results(limit=limit)
@@ -542,11 +556,11 @@ def get_scenario_run_service() -> ScenarioRunService:
     if _service_instance is not None:
         return _service_instance
 
-    max_runs = _DEFAULT_MAX_CONCURRENT_RUNS
+    max_runs = ScenarioRunService.DEFAULT_MAX_CONCURRENT_RUNS
     try:
         from pyrit.backend.main import app
 
-        max_runs = getattr(app.state, "max_concurrent_scenario_runs", _DEFAULT_MAX_CONCURRENT_RUNS)
+        max_runs = getattr(app.state, "max_concurrent_scenario_runs", ScenarioRunService.DEFAULT_MAX_CONCURRENT_RUNS)
     except Exception:
         pass
 

pyrit/backend/services/scenario_service.py

@@ -9,6 +9,7 @@
 """
 
 from functools import lru_cache
+from typing import ClassVar
 
 from pyrit.backend.models.common import PaginationInfo
 from pyrit.backend.models.scenarios import (
@@ -59,26 +60,32 @@ class ScenarioService:
     Uses ScenarioRegistry as the source of truth for scenario metadata.
     """
 
+    DEFAULT_LIST_LIMIT: ClassVar[int] = 50
+
     def __init__(self) -> None:
         """Initialize the scenario service."""
         self._registry = ScenarioRegistry.get_registry_singleton()
 
     async def list_scenarios_async(
         self,
         *,
-        limit: int = 50,
+        limit: int | None = None,
         cursor: str | None = None,
     ) -> ListRegisteredScenariosResponse:
         """
         List all available scenarios with pagination.
 
         Args:
-            limit: Maximum items to return per page.
+            limit: Maximum items to return per page. Defaults to
+                ``DEFAULT_LIST_LIMIT`` (50) when ``None``.
             cursor: Pagination cursor (scenario_name to start after).
 
         Returns:
             ScenarioListResponse with paginated scenario summaries.
         """
+        if limit is None:
+            limit = self.DEFAULT_LIST_LIMIT
+
         all_metadata = self._registry.list_metadata()
         all_summaries = [_metadata_to_registered_scenario(m) for m in all_metadata]
 

pyrit/backend/services/target_service.py

@@ -15,7 +15,7 @@
 import logging
 import os
 from functools import lru_cache
-from typing import Any
+from typing import Any, ClassVar
 from urllib.parse import urlparse
 
 from pyrit import prompt_target
@@ -139,6 +139,8 @@ class TargetService:
     API metadata is derived from the target objects' identifiers.
     """
 
+    DEFAULT_LIST_LIMIT: ClassVar[int] = 50
+
     def __init__(self) -> None:
         """Initialize the target service."""
         self._registry = TargetRegistry.get_registry_singleton()
@@ -177,19 +179,23 @@ def _build_instance_from_object(self, *, target_registry_name: str, target_obj:
     async def list_targets_async(
         self,
         *,
-        limit: int = 50,
+        limit: int | None = None,
         cursor: str | None = None,
     ) -> TargetListResponse:
         """
         List all target instances with pagination.
 
         Args:
-            limit: Maximum items to return.
+            limit: Maximum items to return. Defaults to ``DEFAULT_LIST_LIMIT``
+                (50) when ``None``.
             cursor: Pagination cursor (target_registry_name to start after).
 
         Returns:
             TargetListResponse containing paginated targets.
         """
+        if limit is None:
+            limit = self.DEFAULT_LIST_LIMIT
+
         items = [
             self._build_instance_from_object(target_registry_name=entry.name, target_obj=entry.instance)
             for entry in self._registry.get_all_instances()

pyrit/cli/api_client.py

@@ -12,7 +12,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
+from typing import Any, ClassVar
 
 _logger = logging.getLogger(__name__)
 
@@ -33,6 +33,10 @@ class PyRITApiClient:
             scenarios = await client.list_scenarios_async()
     """
 
+    DEFAULT_REQUEST_TIMEOUT: ClassVar[float] = 60.0
+    DEFAULT_LIST_LIMIT: ClassVar[int] = 200
+    DEFAULT_SCENARIO_RUNS_LIST_LIMIT: ClassVar[int] = 100
+
     def __init__(self, *, base_url: str, request_timeout: float | None = None) -> None:
         """
         Initialize the API client.
@@ -43,10 +47,11 @@ def __init__(self, *, base_url: str, request_timeout: float | None = None) -> No
                 non-polling request (catalog, results, cancel, start, etc.). Polling
                 the live scenario-run endpoint always uses ``read=None`` regardless
                 of this value, because the server may legitimately take many seconds
-                to respond while a scenario is executing. Defaults to ``60.0``.
+                to respond while a scenario is executing. Defaults to
+                ``DEFAULT_REQUEST_TIMEOUT`` (60.0) when ``None``.
         """
         self._base_url = base_url.rstrip("/")
-        self._request_timeout = request_timeout if request_timeout is not None else 60.0
+        self._request_timeout = request_timeout if request_timeout is not None else self.DEFAULT_REQUEST_TIMEOUT
         self._client: Any = None  # httpx.AsyncClient (typed Any to avoid top-level import)
 
     async def __aenter__(self) -> PyRITApiClient:
@@ -92,13 +97,19 @@ async def health_check_async(self) -> bool:
     # Scenarios
     # ------------------------------------------------------------------
 
-    async def list_scenarios_async(self, *, limit: int = 200) -> dict[str, Any]:
+    async def list_scenarios_async(self, *, limit: int | None = None) -> dict[str, Any]:
         """
         List all available scenarios.
 
+        Args:
+            limit: Maximum items to return. Defaults to ``DEFAULT_LIST_LIMIT``
+                (200) when ``None``.
+
         Returns:
             dict: ``ListRegisteredScenariosResponse`` payload.
         """
+        if limit is None:
+            limit = self.DEFAULT_LIST_LIMIT
         return await self._get_json_async(path="/api/scenarios/catalog", params={"limit": limit})
 
     async def get_scenario_async(self, *, scenario_name: str) -> dict[str, Any] | None:
@@ -124,13 +135,19 @@ async def get_scenario_async(self, *, scenario_name: str) -> dict[str, Any] | No
     # Initializers
     # ------------------------------------------------------------------
 
-    async def list_initializers_async(self, *, limit: int = 200) -> dict[str, Any]:
+    async def list_initializers_async(self, *, limit: int | None = None) -> dict[str, Any]:
         """
         List all available initializers.
 
+        Args:
+            limit: Maximum items to return. Defaults to ``DEFAULT_LIST_LIMIT``
+                (200) when ``None``.
+
         Returns:
             dict: ``ListRegisteredInitializersResponse`` payload.
         """
+        if limit is None:
+            limit = self.DEFAULT_LIST_LIMIT
         return await self._get_json_async(path="/api/initializers", params={"limit": limit})
 
     async def register_initializer_async(self, *, name: str, script_content: str) -> dict[str, Any]:
@@ -162,13 +179,19 @@ async def register_initializer_async(self, *, name: str, script_content: str) ->
     # Targets
     # ------------------------------------------------------------------
 
-    async def list_targets_async(self, *, limit: int = 200) -> dict[str, Any]:
+    async def list_targets_async(self, *, limit: int | None = None) -> dict[str, Any]:
         """
         List all available targets.
 
+        Args:
+            limit: Maximum items to return. Defaults to ``DEFAULT_LIST_LIMIT``
+                (200) when ``None``.
+
         Returns:
             dict: ``TargetListResponse`` payload.
         """
+        if limit is None:
+            limit = self.DEFAULT_LIST_LIMIT
         return await self._get_json_async(path="/api/targets", params={"limit": limit})
 
     # ------------------------------------------------------------------
@@ -244,13 +267,19 @@ async def cancel_scenario_run_async(self, *, scenario_result_id: str) -> dict[st
         self._raise_for_status(resp)
         return resp.json()
 
-    async def list_scenario_runs_async(self, *, limit: int = 100) -> dict[str, Any]:
+    async def list_scenario_runs_async(self, *, limit: int | None = None) -> dict[str, Any]:
         """
         List tracked scenario runs.
 
+        Args:
+            limit: Maximum items to return. Defaults to
+                ``DEFAULT_SCENARIO_RUNS_LIST_LIMIT`` (100) when ``None``.
+
         Returns:
             dict: ``ScenarioRunListResponse`` payload.
         """
+        if limit is None:
+            limit = self.DEFAULT_SCENARIO_RUNS_LIST_LIMIT
         return await self._get_json_async(path="/api/scenarios/runs", params={"limit": limit})
 
     # ------------------------------------------------------------------

tests/unit/backend/test_scenario_run_service.py

@@ -17,7 +17,6 @@
     ScenarioRunStatus,
 )
 from pyrit.backend.services.scenario_run_service import (
-    _DEFAULT_MAX_CONCURRENT_RUNS,
     ScenarioRunService,
 )
 from pyrit.models import AttackOutcome
@@ -431,7 +430,7 @@ async def _set_unique_id(**kwargs: object) -> None:
         scenario_instance.initialize_async = AsyncMock(side_effect=_set_unique_id)
 
         # Fill up to the limit
-        for _ in range(_DEFAULT_MAX_CONCURRENT_RUNS):
+        for _ in range(ScenarioRunService.DEFAULT_MAX_CONCURRENT_RUNS):
             await service.start_run_async(request=_make_request())
 
         # Next one should fail