From f5de1f908533e4e640ecd0d62ed38cf8bac246e1 Mon Sep 17 00:00:00 2001 From: Petr Date: Wed, 22 Jul 2026 10:39:22 +0200 Subject: [PATCH] refactor(services): return ResolvedProjectCredentials, not a (stack_url, token) tuple (#516 NB-1) The single-project services (token / stream / snapshot) each carried a byte-identical `_resolve_project(alias) -> tuple[str, str]` returning (stack_url, token) -- a bare heterogeneous 2-tuple whose positional order a caller has to remember. CONTRIBUTING.md ("Return values -- name them with dataclasses, not tuples") forbids this for new code; PR #516 review flagged the new snapshot_service instance (NB-1), and fixing the trio together keeps them consistent rather than diverging from the two grandfathered siblings. - New `ResolvedProjectCredentials` frozen dataclass + shared `resolve_project_credentials(config_store, alias)` helper in services/base.py (the existing home for cross-service free functions like default_client_factory), which also removes the previously-triplicated lookup body. - All three `_resolve_project` helpers delegate to it; call sites now read `creds.stack_url` / `creds.token` instead of positional unpacking. - The "not registered" ConfigError message is preserved verbatim (not switched to the richer ConfigStore.project_not_found_error) so behavior is identical. - Drops the now-unused ConfigError import from token_service. make check clean (4648 passed, 8 skipped); no behavior change. --- src/keboola_agent_cli/services/base.py | 34 ++++++++++++++++++- .../services/snapshot_service.py | 32 ++++++++--------- .../services/stream_service.py | 31 ++++++++--------- .../services/token_service.py | 25 ++++++-------- 4 files changed, 71 insertions(+), 51 deletions(-) diff --git a/src/keboola_agent_cli/services/base.py b/src/keboola_agent_cli/services/base.py index 346ebfa2..60dcd526 100644 --- a/src/keboola_agent_cli/services/base.py +++ b/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. diff --git a/src/keboola_agent_cli/services/snapshot_service.py b/src/keboola_agent_cli/services/snapshot_service.py index 1db3f7b2..871fc4f2 100644 --- a/src/keboola_agent_cli/services/snapshot_service.py +++ b/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) diff --git a/src/keboola_agent_cli/services/stream_service.py b/src/keboola_agent_cli/services/stream_service.py index c6839545..371628fc 100644 --- a/src/keboola_agent_cli/services/stream_service.py +++ b/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: diff --git a/src/keboola_agent_cli/services/token_service.py b/src/keboola_agent_cli/services/token_service.py index 745d62ad..b155f763 100644 --- a/src/keboola_agent_cli/services/token_service.py +++ b/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)