refactor(services): ResolvedProjectCredentials dataclass over (stack_url, token) tuple (#516 NB-1)

src/keboola_agent_cli/services/base.py

@@ -8,10 +8,11 @@
 import os
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass
 from typing import Any
 
 from ..client import KeboolaClient
-from ..config_store import ConfigStore, project_not_found_error
+from ..config_store import ConfigError, ConfigStore, project_not_found_error
 from ..constants import ENV_MAX_PARALLEL_WORKERS, UNEXPECTED_ERROR_MAX_MESSAGE_LEN
 from ..models import ProjectConfig
 
@@ -20,6 +21,37 @@
 ClientFactory = Callable[[str, str], KeboolaClient]
 
 
+@dataclass(frozen=True)
+class ResolvedProjectCredentials:
+    """A project alias resolved to the ``(stack_url, token)`` a client needs.
+
+    Returned by :func:`resolve_project_credentials` and the single-project
+    services' ``_resolve_project`` helpers instead of a bare 2-tuple, so call
+    sites read ``creds.stack_url`` / ``creds.token`` rather than relying on
+    positional order (CONTRIBUTING.md "name them with dataclasses, not tuples").
+    """
+
+    stack_url: str
+    token: str
+
+
+def resolve_project_credentials(
+    config_store: ConfigStore, alias: str
+) -> ResolvedProjectCredentials:
+    """Resolve ``alias`` to its stack URL + token, or raise :class:`ConfigError`.
+
+    Shared by the single-project services (``token`` / ``stream`` / ``snapshot``)
+    whose ``_resolve_project`` helpers were previously byte-identical. Uses the
+    same short, actionable "not registered" message they already emitted (kept
+    verbatim rather than switching to the richer
+    :meth:`ConfigStore.project_not_found_error`, to preserve behavior).
+    """
+    project = config_store.get_project(alias)
+    if project is None:
+        raise ConfigError(f"Project alias '{alias}' is not registered. Run `kbagent project list`.")
+    return ResolvedProjectCredentials(stack_url=project.stack_url, token=project.token)
+
+
 def sanitize_unexpected_error(exc: BaseException) -> str:
     """Truncate an exception message to a safe length for JSON error envelopes.
 

src/keboola_agent_cli/services/snapshot_service.py

@@ -31,6 +31,7 @@
 from ..client import KeboolaClient
 from ..config_store import ConfigStore
 from ..errors import ConfigError, KeboolaApiError
+from .base import ResolvedProjectCredentials, resolve_project_credentials
 
 logger = logging.getLogger(__name__)
 
@@ -66,8 +67,8 @@ def create_snapshot(
         Returns:
             Dict with 'project_alias', 'table_id', 'branch_id', 'snapshot_id'.
         """
-        stack_url, token = self._resolve_project(alias)
-        client = self._client_factory(stack_url, token)
+        creds = self._resolve_project(alias)
+        client = self._client_factory(creds.stack_url, creds.token)
         try:
             results = client.create_table_snapshot(
                 table_id=table_id,
@@ -98,8 +99,8 @@ def list_snapshots(
             Dict with 'project_alias', 'table_id', 'branch_id', 'count',
             'snapshots' (raw API snapshot dicts).
         """
-        stack_url, token = self._resolve_project(alias)
-        client = self._client_factory(stack_url, token)
+        creds = self._resolve_project(alias)
+        client = self._client_factory(creds.stack_url, creds.token)
         try:
             snapshots = client.list_table_snapshots(
                 table_id=table_id,
@@ -122,8 +123,8 @@ def get_snapshot(self, *, alias: str, snapshot_id: str) -> dict[str, Any]:
         Returns:
             Dict with 'project_alias', 'snapshot' (raw API snapshot dict).
         """
-        stack_url, token = self._resolve_project(alias)
-        client = self._client_factory(stack_url, token)
+        creds = self._resolve_project(alias)
+        client = self._client_factory(creds.stack_url, creds.token)
         try:
             snapshot = client.get_snapshot(snapshot_id)
         finally:
@@ -143,13 +144,13 @@ def delete_snapshots(
             Dict with 'project_alias', 'deleted', 'failed', 'dry_run' (and
             'would_delete' when dry_run).
         """
-        stack_url, token = self._resolve_project(alias)
+        creds = self._resolve_project(alias)
 
         deleted: list[str] = []
         failed: list[dict[str, Any]] = []
         would_delete: list[str] = []
 
-        client = self._client_factory(stack_url, token)
+        client = self._client_factory(creds.stack_url, creds.token)
         try:
             for sid in snapshot_ids:
                 if dry_run:
@@ -203,7 +204,7 @@ def create_table_from_snapshot(
         if not name.strip():
             raise ConfigError("table-from-snapshot requires a non-empty --name.")
 
-        stack_url, token = self._resolve_project(alias)
+        creds = self._resolve_project(alias)
 
         if dry_run:
             return {
@@ -215,7 +216,7 @@ def create_table_from_snapshot(
                 "dry_run": True,
             }
 
-        client = self._client_factory(stack_url, token)
+        client = self._client_factory(creds.stack_url, creds.token)
         try:
             table = client.create_table_from_snapshot(
                 bucket_id=bucket_id,
@@ -237,11 +238,6 @@ def create_table_from_snapshot(
             "table_id": table.get("id"),
         }
 
-    def _resolve_project(self, alias: str) -> tuple[str, str]:
-        """Resolve ``alias`` to its ``(stack_url, token)``."""
-        project = self._config_store.get_project(alias)
-        if project is None:
-            raise ConfigError(
-                f"Project alias '{alias}' is not registered. Run `kbagent project list`."
-            )
-        return project.stack_url, project.token
+    def _resolve_project(self, alias: str) -> ResolvedProjectCredentials:
+        """Resolve ``alias`` to its stack URL + token (or raise ConfigError)."""
+        return resolve_project_credentials(self._config_store, alias)

src/keboola_agent_cli/services/stream_service.py

@@ -3,7 +3,8 @@
 Business logic for the ``kbagent stream`` command group. Wraps the Stream
 control-plane API behind a layer that:
 
-- resolves a kbagent project alias to its ``(stack_url, token)`` via
+- resolves a kbagent project alias to a
+  :class:`~keboola_agent_cli.services.base.ResolvedProjectCredentials` via
   :class:`ConfigStore` (the alias is the only handle a caller needs);
 - builds a :class:`StreamClient` through an injectable factory (testability);
 - assembles a source's full picture for ``stream detail`` -- base + per-signal
@@ -32,6 +33,7 @@
 )
 from ..errors import ConfigError, ErrorCode, KeboolaApiError
 from ..stream_client import StreamClient, provision_otlp_sinks, stream_task_source_id
+from .base import ResolvedProjectCredentials, resolve_project_credentials
 
 logger = logging.getLogger(__name__)
 
@@ -66,9 +68,9 @@ def __init__(
 
     def list_sources(self, *, alias: str, branch_id: str | None = None) -> dict[str, Any]:
         """List sources in the alias's project (default branch unless overridden)."""
-        stack_url, token = self._resolve_project(alias)
+        creds = self._resolve_project(alias)
         branch = branch_id or STREAM_DEFAULT_BRANCH
-        client = self._stream_client_factory(stack_url, token)
+        client = self._stream_client_factory(creds.stack_url, creds.token)
         try:
             raw = client.list_sources(branch)
             return {
@@ -101,9 +103,9 @@ def create_source(
         sourceId) is returned untouched with ``status="skipped"`` (its sinks are
         also reconciled when ``provision_sinks`` so a half-set-up source heals).
         """
-        stack_url, token = self._resolve_project(alias)
+        creds = self._resolve_project(alias)
         branch = branch_id or STREAM_DEFAULT_BRANCH
-        client = self._stream_client_factory(stack_url, token)
+        client = self._stream_client_factory(creds.stack_url, creds.token)
         try:
             if if_not_exists:
                 existing = self._find_source(client, branch, name)
@@ -147,9 +149,9 @@ def get_source_detail(
         """Assemble the full picture for one source (endpoints + destination)."""
         if not source_id and not name:
             raise ConfigError("Provide a source id (positional) or --name.")
-        stack_url, token = self._resolve_project(alias)
+        creds = self._resolve_project(alias)
         branch = branch_id or STREAM_DEFAULT_BRANCH
-        client = self._stream_client_factory(stack_url, token)
+        client = self._stream_client_factory(creds.stack_url, creds.token)
         try:
             if source_id:
                 source = client.get_source(branch, source_id)
@@ -176,7 +178,7 @@ def delete_source(
         dry_run: bool = False,
     ) -> dict[str, Any]:
         """Delete a source (async task polled to completion)."""
-        stack_url, token = self._resolve_project(alias)
+        creds = self._resolve_project(alias)
         branch = branch_id or STREAM_DEFAULT_BRANCH
         if dry_run:
             return {
@@ -185,7 +187,7 @@ def delete_source(
                 "branch_id": branch,
                 "source_id": source_id,
             }
-        client = self._stream_client_factory(stack_url, token)
+        client = self._stream_client_factory(creds.stack_url, creds.token)
         try:
             task = client.delete_source(branch, source_id)
             client.wait_for_task(task)
@@ -202,14 +204,9 @@ def delete_source(
     # Internal helpers
     # ------------------------------------------------------------------
 
-    def _resolve_project(self, alias: str) -> tuple[str, str]:
-        """Resolve ``alias`` to its ``(stack_url, token)``."""
-        project = self._config_store.get_project(alias)
-        if project is None:
-            raise ConfigError(
-                f"Project alias '{alias}' is not registered. Run `kbagent project list`."
-            )
-        return project.stack_url, project.token
+    def _resolve_project(self, alias: str) -> ResolvedProjectCredentials:
+        """Resolve ``alias`` to its stack URL + token (or raise ConfigError)."""
+        return resolve_project_credentials(self._config_store, alias)
 
     @staticmethod
     def _find_source(client: StreamClient, branch: str, needle: str) -> dict[str, Any] | None:

src/keboola_agent_cli/services/token_service.py

@@ -23,7 +23,7 @@
 
 from ..client import KeboolaClient
 from ..config_store import ConfigStore
-from ..errors import ConfigError
+from .base import ResolvedProjectCredentials, resolve_project_credentials
 
 logger = logging.getLogger(__name__)
 
@@ -64,14 +64,14 @@ def create_scoped_token(
         API response (its ``token`` field is a one-time secret reveal) plus the
         resolving ``alias``.
         """
-        stack_url, token = self._resolve_project(alias)
+        creds = self._resolve_project(alias)
         bucket_permissions: dict[str, str] = {}
         for bucket_id in bucket_read or []:
             bucket_permissions[bucket_id] = "read"
         for bucket_id in bucket_write or []:
             # write is the stronger grant -- it wins over a read on the same bucket.
             bucket_permissions[bucket_id] = "write"
-        client = self._client_factory(stack_url, token)
+        client = self._client_factory(creds.stack_url, creds.token)
         try:
             result = client.create_scoped_token(
                 description=description,
@@ -86,8 +86,8 @@ def create_scoped_token(
 
     def delete_token(self, *, alias: str, token_id: str) -> dict[str, Any]:
         """Revoke a token immediately in ``alias``'s project."""
-        stack_url, token = self._resolve_project(alias)
-        client = self._client_factory(stack_url, token)
+        creds = self._resolve_project(alias)
+        client = self._client_factory(creds.stack_url, creds.token)
         try:
             client.delete_token(token_id)
             return {"status": "deleted", "alias": alias, "token_id": token_id}
@@ -96,19 +96,14 @@ def delete_token(self, *, alias: str, token_id: str) -> dict[str, Any]:
 
     def refresh_token(self, *, alias: str, token_id: str) -> dict[str, Any]:
         """Rotate a token (old value invalidated) in ``alias``'s project."""
-        stack_url, token = self._resolve_project(alias)
-        client = self._client_factory(stack_url, token)
+        creds = self._resolve_project(alias)
+        client = self._client_factory(creds.stack_url, creds.token)
         try:
             result = client.refresh_token(token_id)
             return {"alias": alias, **result}
         finally:
             client.close()
 
-    def _resolve_project(self, alias: str) -> tuple[str, str]:
-        """Resolve ``alias`` to its ``(stack_url, token)``."""
-        project = self._config_store.get_project(alias)
-        if project is None:
-            raise ConfigError(
-                f"Project alias '{alias}' is not registered. Run `kbagent project list`."
-            )
-        return project.stack_url, project.token
+    def _resolve_project(self, alias: str) -> ResolvedProjectCredentials:
+        """Resolve ``alias`` to its stack URL + token (or raise ConfigError)."""
+        return resolve_project_credentials(self._config_store, alias)