Phase 2: Foundation — Shared Constants & HTTP Base

src/keboola_agent_cli/client.py

@@ -3,43 +3,42 @@
 This is the only module that communicates with the Keboola Storage API
 and the Keboola Queue API. All HTTP details, endpoint URLs, and error
 mapping are encapsulated here.
+
+Inherits shared retry/error logic from BaseHttpClient.
 """
 
-import time
 from typing import Any
 from urllib.parse import quote, urlparse, urlunparse
 
 import httpx
 
 from . import __version__
-from .errors import KeboolaApiError, mask_token
+from .constants import DEFAULT_JOB_LIMIT, DEFAULT_TIMEOUT
+from .http_base import BaseHttpClient
 from .models import TokenVerifyResponse
 
-RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
-MAX_RETRIES = 3
-BACKOFF_BASE = 1.0  # seconds; delays: 1s, 2s, 4s
-
 
-class KeboolaClient:
+class KeboolaClient(BaseHttpClient):
     """HTTP client for the Keboola Storage API and Queue API.
 
     Provides methods to interact with Keboola endpoints with built-in
     retry logic (exponential backoff for 429/5xx), timeouts, and
     automatic token masking in error messages.
+
+    Inherits _do_request() and _raise_api_error() from BaseHttpClient.
     """
 
     def __init__(self, stack_url: str, token: str) -> None:
         self._stack_url = stack_url.rstrip("/")
-        self._token = token
-        self._masked_token = mask_token(token)
-        self._headers = {
+        headers = {
             "X-StorageApi-Token": token,
             "User-Agent": f"keboola-agent-cli/{__version__}",
         }
-        self._client = httpx.Client(
+        super().__init__(
             base_url=self._stack_url,
-            timeout=httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0),
-            headers=self._headers,
+            token=token,
+            headers=headers,
+            timeout=DEFAULT_TIMEOUT,
         )
         self._queue_client: httpx.Client | None = None
 
@@ -57,7 +56,7 @@ def _queue_base_url(self) -> str:
 
     def close(self) -> None:
         """Close the underlying HTTP clients."""
-        self._client.close()
+        super().close()
         if self._queue_client is not None:
             self._queue_client.close()
 
@@ -67,132 +66,23 @@ def __enter__(self) -> "KeboolaClient":
     def __exit__(self, *args: Any) -> None:
         self.close()
 
-    def _do_request(
-        self, client: httpx.Client, base_url: str, method: str, path: str, **kwargs: Any
-    ) -> httpx.Response:
-        """Execute an HTTP request with retry and exponential backoff.
-
-        Shared retry logic for both Storage and Queue API clients.
-
-        Retries on status codes 429, 500, 502, 503, 504 up to MAX_RETRIES times
-        with exponential backoff (1s, 2s, 4s).
-
-        Raises:
-            KeboolaApiError: On HTTP errors (with masked token) or after retries exhausted.
-        """
-        last_response: httpx.Response | None = None
-
-        for attempt in range(MAX_RETRIES):
-            try:
-                response = client.request(method, path, **kwargs)
-
-                if response.status_code < 400:
-                    return response
-
-                if response.status_code in RETRYABLE_STATUS_CODES and attempt < MAX_RETRIES - 1:
-                    delay = BACKOFF_BASE * (2**attempt)
-                    time.sleep(delay)
-                    last_response = response
-                    continue
-
-                self._raise_api_error(response, base_url)
-
-            except httpx.TimeoutException as exc:
-                if attempt < MAX_RETRIES - 1:
-                    delay = BACKOFF_BASE * (2**attempt)
-                    time.sleep(delay)
-                    continue
-                raise KeboolaApiError(
-                    message=f"Request timed out connecting to {base_url} (token: {self._masked_token})",
-                    status_code=0,
-                    error_code="TIMEOUT",
-                    retryable=True,
-                ) from exc
-
-            except httpx.ConnectError as exc:
-                if attempt < MAX_RETRIES - 1:
-                    delay = BACKOFF_BASE * (2**attempt)
-                    time.sleep(delay)
-                    continue
-                raise KeboolaApiError(
-                    message=f"Cannot connect to {base_url} (token: {self._masked_token})",
-                    status_code=0,
-                    error_code="CONNECTION_ERROR",
-                    retryable=True,
-                ) from exc
-
-        if last_response is not None:
-            self._raise_api_error(last_response, base_url)
-
-        raise KeboolaApiError(
-            message=f"Request failed after {MAX_RETRIES} retries to {base_url} (token: {self._masked_token})",
-            status_code=0,
-            error_code="RETRY_EXHAUSTED",
-            retryable=True,
-        )
-
     def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
         """Execute a Storage API request with retry."""
-        return self._do_request(self._client, self._stack_url, method, path, **kwargs)
+        return self._do_request(method, path, **kwargs)
 
     def _queue_request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
         """Execute a Queue API request with retry. Lazily creates the queue client."""
         if self._queue_client is None:
             self._queue_client = httpx.Client(
                 base_url=self._queue_base_url,
-                timeout=httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0),
-                headers=self._headers,
+                timeout=DEFAULT_TIMEOUT,
+                headers=self._client._headers.copy(),
             )
         return self._do_request(
-            self._queue_client, self._queue_base_url, method, path, **kwargs
-        )
-
-    def _raise_api_error(self, response: httpx.Response, base_url: str | None = None) -> None:
-        """Convert an HTTP error response into a KeboolaApiError."""
-        status = response.status_code
-        url_label = base_url or self._stack_url
-
-        try:
-            body = response.json()
-            api_message = body.get("error", body.get("message", response.text))
-        except Exception:
-            api_message = response.text
-
-        # Truncate to prevent Rich markup injection and excessive output
-        max_api_error_length = 500
-        if isinstance(api_message, str) and len(api_message) > max_api_error_length:
-            api_message = api_message[:max_api_error_length] + "..."
-
-        if status == 401:
-            raise KeboolaApiError(
-                message=f"Invalid or expired token (token: {self._masked_token}): {api_message}",
-                status_code=status,
-                error_code="INVALID_TOKEN",
-                retryable=False,
-            )
-
-        if status == 403:
-            raise KeboolaApiError(
-                message=f"Access denied (token: {self._masked_token}): {api_message}",
-                status_code=status,
-                error_code="ACCESS_DENIED",
-                retryable=False,
-            )
-
-        if status == 404:
-            raise KeboolaApiError(
-                message=f"Resource not found: {api_message}",
-                status_code=status,
-                error_code="NOT_FOUND",
-                retryable=False,
-            )
-
-        retryable = status in RETRYABLE_STATUS_CODES
-        raise KeboolaApiError(
-            message=f"API error {status} from {url_label} (token: {self._masked_token}): {api_message}",
-            status_code=status,
-            error_code="API_ERROR",
-            retryable=retryable,
+            method, path,
+            client=self._queue_client,
+            base_url=self._queue_base_url,
+            **kwargs,
         )
 
     def verify_token(self) -> TokenVerifyResponse:
@@ -269,7 +159,7 @@ def list_jobs(
         component_id: str | None = None,
         config_id: str | None = None,
         status: str | None = None,
-        limit: int = 50,
+        limit: int = DEFAULT_JOB_LIMIT,
         offset: int = 0,
     ) -> list[dict[str, Any]]:
         """List jobs from the Queue API.
@@ -278,7 +168,7 @@ def list_jobs(
             component_id: Optional filter by component ID.
             config_id: Optional filter by config ID (requires component_id).
             status: Optional filter by job status.
-            limit: Max number of jobs to return (1-500, default 50).
+            limit: Max number of jobs to return (1-500).
             offset: Offset for pagination.
 
         Returns:

src/keboola_agent_cli/commands/job.py

@@ -6,6 +6,7 @@
 
 import typer
 
+from ..constants import DEFAULT_JOB_LIMIT, MAX_JOB_LIMIT
 from ..errors import ConfigError, KeboolaApiError
 from ..output import OutputFormatter, format_job_detail, format_jobs_table
 from ..services.job_service import JobService
@@ -49,9 +50,9 @@ def job_list(
         help="Filter by job status: processing, terminated, cancelled, success, error",
     ),
     limit: int = typer.Option(
-        50,
+        DEFAULT_JOB_LIMIT,
         "--limit",
-        help="Maximum number of jobs to return per project (1-500)",
+        help=f"Maximum number of jobs to return per project (1-{MAX_JOB_LIMIT})",
     ),
 ) -> None:
     """List jobs from connected projects."""
@@ -67,9 +68,9 @@ def job_list(
         raise typer.Exit(code=2)
 
     # Validate limit range
-    if limit < 1 or limit > 500:
+    if limit < 1 or limit > MAX_JOB_LIMIT:
         formatter.error(
-            message=f"Invalid limit {limit}. Must be between 1 and 500.",
+            message=f"Invalid limit {limit}. Must be between 1 and {MAX_JOB_LIMIT}.",
             error_code="INVALID_ARGUMENT",
         )
         raise typer.Exit(code=2)

src/keboola_agent_cli/commands/org.py

@@ -11,6 +11,7 @@
 from rich.console import Console
 from rich.table import Table
 
+from ..constants import DEFAULT_TOKEN_DESCRIPTION, ENV_KBC_MANAGE_API_TOKEN, ENV_KBC_STORAGE_API_URL
 from ..errors import KeboolaApiError
 from ..output import OutputFormatter
 from ..services.org_service import OrgService
@@ -42,7 +43,7 @@ def _resolve_manage_token() -> str:
     Raises:
         typer.Exit: If no token can be resolved.
     """
-    env_token = os.environ.get("KBC_MANAGE_API_TOKEN")
+    env_token = os.environ.get(ENV_KBC_MANAGE_API_TOKEN)
     if env_token:
         return env_token
 
@@ -51,7 +52,7 @@ def _resolve_manage_token() -> str:
         return typer.prompt("Manage API token", hide_input=True)
 
     typer.echo(
-        "Error: No manage token available. Set KBC_MANAGE_API_TOKEN env var "
+        f"Error: No manage token available. Set {ENV_KBC_MANAGE_API_TOKEN} env var "
         "or run interactively.",
         err=True,
     )
@@ -141,7 +142,7 @@ def org_setup(
     url: str = typer.Option(
         ...,
         "--url",
-        envvar="KBC_STORAGE_API_URL",
+        envvar=ENV_KBC_STORAGE_API_URL,
         help="Keboola stack URL (e.g. https://connection.keboola.com)",
     ),
     dry_run: bool = typer.Option(
@@ -156,7 +157,7 @@ def org_setup(
         help="Skip confirmation prompt",
     ),
     token_description: str = typer.Option(
-        "kbagent-cli",
+        DEFAULT_TOKEN_DESCRIPTION,
         "--token-description",
         help="Description prefix for created Storage API tokens",
     ),

src/keboola_agent_cli/commands/project.py

@@ -12,6 +12,7 @@
 from rich.console import Console
 from rich.table import Table
 
+from ..constants import DEFAULT_STACK_URL, ENV_KBC_STORAGE_API_URL, ENV_KBC_TOKEN
 from ..errors import ConfigError, KeboolaApiError
 from ..output import OutputFormatter
 from ..services.project_service import ProjectService
@@ -101,7 +102,7 @@ def _resolve_token() -> str:
     Raises:
         typer.Exit: If no token can be resolved.
     """
-    env_token = os.environ.get("KBC_TOKEN")
+    env_token = os.environ.get(ENV_KBC_TOKEN)
     if env_token:
         return env_token
 
@@ -110,7 +111,7 @@ def _resolve_token() -> str:
         return typer.prompt("Storage API token", hide_input=True)
 
     typer.echo(
-        "Error: No token available. Set KBC_TOKEN env var "
+        f"Error: No token available. Set {ENV_KBC_TOKEN} env var "
         "or run interactively.",
         err=True,
     )
@@ -122,9 +123,9 @@ def project_add(
     ctx: typer.Context,
     alias: str = typer.Option(..., help="Human-friendly name for this project"),
     url: str = typer.Option(
-        "https://connection.keboola.com",
+        DEFAULT_STACK_URL,
         help="Keboola stack URL",
-        envvar="KBC_STORAGE_API_URL",
+        envvar=ENV_KBC_STORAGE_API_URL,
     ),
 ) -> None:
     """Add a new Keboola project connection.

src/keboola_agent_cli/constants.py

@@ -0,0 +1,37 @@
+"""Shared constants for Keboola Agent CLI.
+
+All magic numbers, default values, retry parameters, timeout settings,
+and environment variable names are centralized here to avoid duplication
+and ensure consistency across the codebase.
+"""
+
+import httpx
+
+# --- HTTP Retry Constants ---
+RETRYABLE_STATUS_CODES: set[int] = {429, 500, 502, 503, 504}
+MAX_RETRIES: int = 3
+BACKOFF_BASE: float = 1.0  # seconds; delays: 1s, 2s, 4s
+
+# --- HTTP Timeout ---
+DEFAULT_TIMEOUT: httpx.Timeout = httpx.Timeout(
+    connect=5.0, read=30.0, write=10.0, pool=5.0
+)
+
+# --- API Error Handling ---
+MAX_API_ERROR_LENGTH: int = 500
+
+# --- Default Stack URL ---
+DEFAULT_STACK_URL: str = "https://connection.keboola.com"
+
+# --- Token Description ---
+DEFAULT_TOKEN_DESCRIPTION: str = "kbagent-cli"
+
+# --- Job Limits ---
+DEFAULT_JOB_LIMIT: int = 50
+MAX_JOB_LIMIT: int = 500
+
+# --- Environment Variable Names ---
+ENV_MAX_PARALLEL_WORKERS: str = "KBAGENT_MAX_PARALLEL_WORKERS"
+ENV_KBC_TOKEN: str = "KBC_TOKEN"
+ENV_KBC_STORAGE_API_URL: str = "KBC_STORAGE_API_URL"
+ENV_KBC_MANAGE_API_TOKEN: str = "KBC_MANAGE_API_TOKEN"