diff --git a/src/trello_plugin/client.py b/src/trello_plugin/client.py index 61796a6..1b166a1 100644 --- a/src/trello_plugin/client.py +++ b/src/trello_plugin/client.py @@ -8,13 +8,47 @@ Provides the TrelloClient for interacting with the Trello REST API. from __future__ import annotations import os -from typing import Any +from typing import Any, TypeGuard, TypedDict import requests +# Base URL for the Trello REST API v1. +# See https://developer.atlassian.com/cloud/trello/rest TRELLO_API_BASE = "https://api.trello.com/1" +# --------------------------------------------------------------------------- +# Typed response contracts +# --------------------------------------------------------------------------- + + +class SuccessResponse(TypedDict): + """A successful API response carrying data.""" + + success: bool + data: dict[str, Any] + + +class ErrorResponse(TypedDict): + """An unsuccessful API response with a human-readable message.""" + + success: bool + message: str + + +ApiResponse = SuccessResponse | ErrorResponse + + +def _is_success(resp: ApiResponse) -> TypeGuard[SuccessResponse]: + """Narrow an ApiResponse union to its success branch.""" + return bool(resp.get("success", False)) # noqa: FBT003 + + +# --------------------------------------------------------------------------- +# Client +# --------------------------------------------------------------------------- + + class TrelloClient: """Client for the Trello REST API. @@ -23,6 +57,10 @@ class TrelloClient: JSON serialization by Hermes tool handlers. """ + api_key: str + token: str + _session: requests.Session + def __init__( self, api_key: str | None = None, @@ -41,13 +79,42 @@ class TrelloClient: """Return query parameters common to every Trello API call.""" return {"key": self.api_key, "token": self.token} - def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: - """Make an authenticated GET request and return the JSON response.""" + def _check_credentials(self) -> ErrorResponse | None: + """Return an ErrorResponse if credentials are missing, else None.""" if not self.api_key or not self.token: - return { - "success": False, - "message": "TRELLO_API_KEY and TRELLO_TOKEN must both be set as environment variables.", - } + return ErrorResponse( + success=False, + message="TRELLO_API_KEY and TRELLO_TOKEN must both be set as environment variables.", + ) + return None + + def _handle_http_error(self, exc: requests.exceptions.HTTPError) -> ErrorResponse: + """Map known HTTP status codes to user-facing error messages.""" + status = exc.response.status_code if exc.response is not None else 0 + messages: dict[int, str] = { + 401: "Trello authentication failed (401). Check your TRELLO_API_KEY and TRELLO_TOKEN.", + 403: "Trello access denied (403). Your token may not have the required scopes.", + 429: "Trello rate limit exceeded. Try again later.", + } + message = messages.get( + status, f"Trello API error ({status}): {exc}" + ) + return ErrorResponse(success=False, message=message) + + def _request( + self, + method: str, + path: str, + params: dict[str, Any] | None = None, + json_body: dict[str, Any] | None = None, + ) -> ApiResponse: + """Make an authenticated HTTP request and return the JSON response. + + Handles credential checks, HTTP errors, network errors, and timeouts. + """ + cred_error = self._check_credentials() + if cred_error is not None: + return cred_error url = f"{TRELLO_API_BASE}{path}" merged_params = self._auth_params() @@ -55,46 +122,46 @@ class TrelloClient: merged_params.update(params) try: - resp = self._session.get(url, params=merged_params, timeout=15) + resp = self._session.request( + method, url, params=merged_params, json=json_body, timeout=15 + ) resp.raise_for_status() data: dict[str, Any] = resp.json() - return {"success": True, "data": data} + return SuccessResponse(success=True, data=data) except requests.exceptions.HTTPError as exc: - status = exc.response.status_code if exc.response is not None else 0 - if status == 401: - return { - "success": False, - "message": "Trello authentication failed (401). Check your TRELLO_API_KEY and TRELLO_TOKEN.", - } - if status == 403: - return { - "success": False, - "message": "Trello access denied (403). Your token may not have the required scopes.", - } - if status == 429: - return { - "success": False, - "message": "Trello rate limit exceeded. Try again later.", - } - return { - "success": False, - "message": f"Trello API error ({status}): {exc}", - } + return self._handle_http_error(exc) except requests.exceptions.ConnectionError: - return { - "success": False, - "message": "Could not connect to Trello API. Check your network connection.", - } + return ErrorResponse( + success=False, + message="Could not connect to Trello API. Check your network connection.", + ) except requests.exceptions.Timeout: - return { - "success": False, - "message": "Trello API request timed out. Try again later.", - } + return ErrorResponse( + success=False, + message="Trello API request timed out. Try again later.", + ) except requests.exceptions.RequestException as exc: - return { - "success": False, - "message": f"Trello API request failed: {exc}", - } + return ErrorResponse( + success=False, message=f"Trello API request failed: {exc}" + ) + + def _get( + self, path: str, params: dict[str, Any] | None = None + ) -> ApiResponse: + """Make an authenticated GET request.""" + return self._request("GET", path, params=params) + + def _post( + self, path: str, json_body: dict[str, Any] | None = None + ) -> ApiResponse: + """Make an authenticated POST request.""" + return self._request("POST", path, json_body=json_body) + + def _put( + self, path: str, json_body: dict[str, Any] | None = None + ) -> ApiResponse: + """Make an authenticated PUT request.""" + return self._request("PUT", path, json_body=json_body) # ------------------------------------------------------------------ # Public API — Auth & Connection @@ -108,7 +175,7 @@ class TrelloClient: ``member`` key contains the Trello member id and username. """ result = self._get("/members/me") - if result["success"]: + if _is_success(result): member = result["data"] return { "success": True, @@ -134,8 +201,10 @@ class TrelloClient: dict with ``success`` bool and a ``boards`` list (each containing ``id``, ``name``, ``url``, ``closed``, ``starred``). """ - result = self._get("/members/me/boards", params={"fields": "id,name,url,closed,starred"}) - if not result["success"]: + result = self._get( + "/members/me/boards", params={"fields": "id,name,url,closed,starred"} + ) + if not _is_success(result): return { "success": False, "message": result.get("message", "Failed to list Trello boards."), @@ -156,6 +225,10 @@ class TrelloClient: def disconnect(self) -> dict[str, Any]: """Clear credentials from the in-memory client. + Since credentials are stored in environment variables, the next tool call + will automatically re-read them and reconnect. To fully disconnect, also + unset TRELLO_API_KEY and TRELLO_TOKEN from the Hermes profile config. + Note: this does NOT revoke the Trello token. The user must invalidate it via Trello's account settings if they want full revocation. """ @@ -164,12 +237,234 @@ class TrelloClient: return { "success": True, "message": ( - "Trello credentials cleared. " - "Set TRELLO_API_KEY and TRELLO_TOKEN again to reconnect." + "Trello credentials cleared from memory. " + "Next tool call will re-read TRELLO_API_KEY and TRELLO_TOKEN " + "from environment and reconnect automatically." ), } + # ------------------------------------------------------------------ + # Public API — Board Management + # ------------------------------------------------------------------ + + def _resolve_board_id(self, board_id: str) -> dict[str, Any]: + """Resolve a board name or ID to a Trello board ID. + + Returns board info dict with at least an ``id`` key on success, + or an error dict with ``success: False``. + """ + # Try as ID first (Trello IDs are 8-24 char hex strings) + result = self._get(f"/boards/{board_id}", params={"fields": "id,name,url"}) + if _is_success(result): + board = result["data"] + return {"id": board["id"], "name": board.get("name", ""), "url": board.get("url", "")} + + # Try resolving by name + boards_result = self.list_boards() + if not _is_success(boards_result): # type: ignore[arg-type] + return { + "success": False, + "message": ( + f"Board '{board_id}' not found. " + "Use trello_list_boards to see available boards." + ), + } + + matches = [b for b in boards_result["boards"] if b["name"] == board_id] # type: ignore[typeddict-item] + if len(matches) == 1: + return dict(matches[0]) # already has id, name, url + + if len(matches) > 1: + return { + "success": False, + "message": ( + f"Multiple boards named '{board_id}' found. " + "Use the board ID instead of the name." + ), + } + + return { + "success": False, + "message": ( + f"Board '{board_id}' not found. " + "Use trello_list_boards to see available boards." + ), + } + + def create_board( + self, name: str, default_lists: bool = True + ) -> dict[str, Any]: + """Create a new Trello board. + + Args: + name: The name for the new board. + default_lists: Whether to create the default lists (default: True). + + Returns: + dict with board details on success, or an error dict. + """ + body: dict[str, Any] = { + "name": name, + "defaultLists": default_lists, + } + result = self._post("/boards", json_body=body) + if _is_success(result): + board = result["data"] + return { + "success": True, + "board": { + "id": board.get("id"), + "name": board.get("name"), + "url": board.get("url"), + }, + } + return { + "success": False, + "message": result.get("message", "Failed to create board."), + } + + def rename_board(self, board_id: str, name: str) -> dict[str, Any]: + """Rename an existing Trello board. + + Args: + board_id: Board ID or name. + name: The new name for the board. + + Returns: + dict with board details on success, or an error dict. + """ + resolved = self._resolve_board_id(board_id) + if "success" in resolved and resolved["success"] is False: + return resolved # type: ignore[typeddict-item] + + trello_id = resolved["id"] + result = self._put( + f"/boards/{trello_id}", json_body={"name": name} + ) + if _is_success(result): + board = result["data"] + return { + "success": True, + "board": { + "id": board.get("id"), + "name": board.get("name"), + "url": board.get("url"), + }, + } + return { + "success": False, + "message": result.get("message", "Failed to rename board."), + } + + def archive_board(self, board_id: str) -> dict[str, Any]: + """Close/archive a Trello board. + + Args: + board_id: Board ID or name. + + Returns: + dict with success message or error. + """ + resolved = self._resolve_board_id(board_id) + if "success" in resolved and resolved["success"] is False: + return resolved # type: ignore[typeddict-item] + + trello_id = resolved["id"] + board_name = resolved.get("name", trello_id) + result = self._put( + f"/boards/{trello_id}", json_body={"closed": True} + ) + if _is_success(result): + return {"success": True, "message": f"Board '{board_name}' archived."} + return { + "success": False, + "message": result.get("message", "Failed to archive board."), + } + + def open_board(self, board_id: str) -> dict[str, Any]: + """Re-open a closed/archived Trello board. + + Args: + board_id: Board ID or name. + + Returns: + dict with success message or error. + """ + resolved = self._resolve_board_id(board_id) + if "success" in resolved and resolved["success"] is False: + return resolved # type: ignore[typeddict-item] + + trello_id = resolved["id"] + board_name = resolved.get("name", trello_id) + result = self._put( + f"/boards/{trello_id}", json_body={"closed": False} + ) + if _is_success(result): + return {"success": True, "message": f"Board '{board_name}' opened."} + return { + "success": False, + "message": result.get("message", "Failed to open board."), + } + + def board_details(self, board_id: str) -> dict[str, Any]: + """Fetch detailed information about a Trello board. + + Returns the board's lists, members, and a direct URL. + + Args: + board_id: Board ID or name. + + Returns: + dict with board details including lists and members on success. + """ + resolved = self._resolve_board_id(board_id) + if "success" in resolved and resolved["success"] is False: + return resolved # type: ignore[typeddict-item] + + trello_id = resolved["id"] + result = self._get( + f"/boards/{trello_id}", + params={ + "fields": "id,name,desc,url,closed,starred", + "lists": "all", + "members": "all", + "members_fields": "id,username,fullName", + }, + ) + if not _is_success(result): + return { + "success": False, + "message": result.get("message", "Failed to fetch board details."), + } + + board = result["data"] + return { + "success": True, + "board": { + "id": board.get("id"), + "name": board.get("name"), + "url": board.get("url"), + "desc": board.get("desc", ""), + "closed": board.get("closed", False), + "starred": board.get("starred", False), + "lists": [ + {"id": lst.get("id"), "name": lst.get("name")} + for lst in board.get("lists", []) + ], + "members": [ + { + "id": m.get("id"), + "username": m.get("username"), + "full_name": m.get("fullName"), + } + for m in board.get("members", []) + ], + }, + } + @staticmethod def check_requirements() -> bool: """Check if the required environment variables are set.""" - return bool(os.environ.get("TRELLO_API_KEY")) and bool(os.environ.get("TRELLO_TOKEN")) \ No newline at end of file + return bool(os.environ.get("TRELLO_API_KEY")) and bool( + os.environ.get("TRELLO_TOKEN") + ) \ No newline at end of file diff --git a/src/trello_plugin/tools.py b/src/trello_plugin/tools.py index 8920923..ab3afe1 100644 --- a/src/trello_plugin/tools.py +++ b/src/trello_plugin/tools.py @@ -73,7 +73,7 @@ def trello_disconnect() -> str: """ global _client # noqa: PLW0603 _client = None - return _respond({"success": True, "message": "Trello credentials cleared. Reconnect by calling trello_verify_credentials after setting TRELLO_API_KEY and TRELLO_TOKEN."}) + return _respond({"success": True, "message": "Trello credentials cleared from memory. Next tool call will re-read TRELLO_API_KEY and TRELLO_TOKEN from environment and reconnect automatically."}) # --------------------------------------------------------------------------- diff --git a/tests/test_auth.py b/tests/test_auth.py index 857a445..b44767b 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -8,6 +8,7 @@ from collections.abc import Generator from typing import Any import pytest +from pytest import MonkeyPatch import requests # Tell pytest we plan to set env vars in tests @@ -19,16 +20,11 @@ pytestmark = pytest.mark.usefixtures("clear_env") # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) -def clear_env() -> Generator[None, None, None]: +def clear_env(monkeypatch: MonkeyPatch) -> Generator[None, None, None]: """Remove Trello env vars before each test so state is predictable.""" - # Use monkeypatch via request directly — fallback to os.environ - saved_key = os.environ.pop("TRELLO_API_KEY", None) - saved_token = os.environ.pop("TRELLO_TOKEN", None) + monkeypatch.delenv("TRELLO_API_KEY", raising=False) + monkeypatch.delenv("TRELLO_TOKEN", raising=False) yield - if saved_key is not None: - os.environ["TRELLO_API_KEY"] = saved_key - if saved_token is not None: - os.environ["TRELLO_TOKEN"] = saved_token def _make_session(mocker: Any) -> requests.Session: