diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 57a967e..8f01e3e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -27,9 +27,9 @@ jobs: - name: Sync deps (frozen) run: | - ~/.cargo/bin/uv venv --python 3.13 + $HOME/.local/bin/uv venv --python 3.12 source .venv/bin/activate - ~/.cargo/bin/uv sync --frozen + $HOME/.local/bin/uv sync --frozen - name: Lint & Test run: | diff --git a/.python-version b/.python-version deleted file mode 100644 index 9c9b206..0000000 --- a/.python-version +++ /dev/null @@ -1 +0,0 @@ ->=3.14 diff --git a/pyproject.toml b/pyproject.toml index f8f9153..d2cf056 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,7 +3,7 @@ name = "prusa-connect" dynamic = ["version"] description = "Prusa Connect CLI and API client" readme = "README.md" -requires-python = ">=3.14" +requires-python = ">=3.12" dependencies = [ "requests>=2.32.5", "structlog>=25.5.0", diff --git a/src/prusa_connect/__init__.py b/src/prusa_connect/__init__.py index ec855d5..2ad6455 100644 --- a/src/prusa_connect/__init__.py +++ b/src/prusa_connect/__init__.py @@ -1,3 +1,19 @@ import importlib.metadata +from .client import PrusaConnectClient +from .exceptions import PrusaApiError, PrusaAuthError, PrusaNetworkError +from .models import Camera, File, Job, Printer, Team + __version__ = importlib.metadata.version("prusa-connect") + +__all__ = [ + "Camera", + "File", + "Job", + "Printer", + "PrusaApiError", + "PrusaAuthError", + "PrusaConnectClient", + "PrusaNetworkError", + "Team", +] diff --git a/src/prusa_connect/auth.py b/src/prusa_connect/auth.py index cd63e12..b63a126 100644 --- a/src/prusa_connect/auth.py +++ b/src/prusa_connect/auth.py @@ -5,7 +5,6 @@ and attach headers to requests. """ -from datetime import timedelta import base64 import hashlib import json @@ -13,7 +12,7 @@ import re import urllib.parse from collections.abc import Callable -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any diff --git a/src/prusa_connect/cli.py b/src/prusa_connect/cli.py index da2aaba..242dd8c 100644 --- a/src/prusa_connect/cli.py +++ b/src/prusa_connect/cli.py @@ -11,7 +11,6 @@ import better_exceptions import cyclopts -import requests import structlog from rich import print as rprint from rich.console import Console @@ -377,10 +376,10 @@ def api( # Use raw=True if output is specified to handle binary raw_mode = output is not None - - # Check if we should default to raw for binary-like paths if not specified? + + # Check if we should default to raw for binary-like paths if not specified? # For now, explicit output flag implies raw. - + try: res = client._request(method, path, raw=raw_mode, **kwargs) diff --git a/src/prusa_connect/client.py b/src/prusa_connect/client.py index 660cbc7..f671635 100644 --- a/src/prusa_connect/client.py +++ b/src/prusa_connect/client.py @@ -19,6 +19,9 @@ logger = structlog.get_logger() DEFAULT_BASE_URL = "https://connect.prusa3d.com/app" +DEFAULT_TIMEOUT = 10.0 + +__all__ = ["PrusaConnectClient"] class AuthStrategy(Protocol): @@ -59,13 +62,21 @@ def __init__(self, credentials: AuthStrategy, base_url: str = DEFAULT_BASE_URL) } ) - def _request(self, method: str, endpoint: str, raw: bool = False, **kwargs: Any) -> Any: + def _request( + self, + method: str, + endpoint: str, + raw: bool = False, + timeout: float = DEFAULT_TIMEOUT, + **kwargs: Any, + ) -> Any: """Internal method to handle requests, errors, and logging. Args: method: HTTP method (GET, POST, etc.). endpoint: API endpoint (e.g., '/printers'). raw: If True, return the raw response object instead of parsing JSON. + timeout: Request timeout in seconds. **kwargs: Additional arguments passed to requests.request. Returns: @@ -84,15 +95,34 @@ def _request(self, method: str, endpoint: str, raw: bool = False, **kwargs: Any) try: logger.debug("API Request", method=method, url=url) - response = self._session.request(method, url, **kwargs) - logger.debug("API Response", status_code=response.status_code, headers=response.headers, body_len=len(response.content)) + response = self._session.request(method, url, timeout=timeout, **kwargs) + logger.debug( + "API Response", + status_code=response.status_code, + headers=response.headers, + body_len=len(response.content), + ) if response.status_code in (401, 403): raise PrusaAuthError("Invalid or expired credentials.") if response.status_code >= 400: + message = f"Request failed: {response.reason}" + # Attempt to parse detailed error message from JSON response + try: + error_json = response.json() + # Common patterns: {"message": "..."}, {"error": "..."}, {"detail": "..."} + if isinstance(error_json, dict): + for key in ("message", "error", "detail"): + if key in error_json and isinstance(error_json[key], str): + message = error_json[key] + break + except ValueError: + # Not JSON, stick to reason + pass + raise PrusaApiError( - message=f"Request failed: {response.reason}", + message=message, status_code=response.status_code, response_body=response.text[:500], ) @@ -109,15 +139,16 @@ def _request(self, method: str, endpoint: str, raw: bool = False, **kwargs: Any) logger.error("Network error", error=str(e)) raise PrusaNetworkError(f"Failed to connect to Prusa Connect: {e}") from e - def api_request(self, method: str, endpoint: str, **kwargs) -> Any: - """Public wrapper for making raw authenticated requests.""" - return self._request(method, endpoint, **kwargs) - def get_printers(self) -> list[Printer]: """Fetch all printers associated with the account. Returns: A list of Printer objects. + + Usage Example: + >>> printers = client.get_printers() + >>> for printer in printers: + ... print(printer.name) """ data = self._request("GET", "/printers") @@ -139,6 +170,10 @@ def get_printer(self, uuid: str) -> Printer: Returns: A Printer object. + + Usage Example: + >>> printer = client.get_printer("c0ffee-1234") + >>> print(printer.printer_state) """ data = self._request("GET", f"/printers/{uuid}") return Printer.model_validate(data) @@ -151,6 +186,11 @@ def get_file_list(self, team_id: int) -> list[File]: Returns: A list of File objects. + + Usage Example: + >>> files = client.get_file_list(team_id=123) + >>> for file in files: + ... print(file.name) """ # Note: The endpoint might vary based on your reverse engineering. # Assuming /teams/{id}/files based on typical Prusa structure or similar. @@ -165,6 +205,11 @@ def get_cameras(self) -> list[Camera]: Returns: A list of Camera objects. + + Usage Example: + >>> cameras = client.get_cameras() + >>> for cam in cameras: + ... print(cam.name) """ data = self._request("GET", "/cameras") if isinstance(data, dict) and "cameras" in data: @@ -176,6 +221,11 @@ def get_teams(self) -> list[Team]: Returns: A list of Team objects. + + Usage Example: + >>> teams = client.get_teams() + >>> for team in teams: + ... print(team.name) """ data = self._request("GET", "/users/teams") if isinstance(data, dict) and "teams" in data: @@ -190,6 +240,11 @@ def get_team_jobs(self, team_id: int) -> list[Job]: Returns: A list of Job objects. + + Usage Example: + >>> jobs = client.get_team_jobs(team_id=123) + >>> for job in jobs: + ... print(job.state) """ data = self._request("GET", f"/teams/{team_id}/jobs") if isinstance(data, dict) and "jobs" in data: @@ -204,6 +259,11 @@ def get_printer_jobs(self, printer_uuid: str) -> list[Job]: Returns: A list of Job objects. + + Usage Example: + >>> jobs = client.get_printer_jobs("c0ffee-1234") + >>> for job in jobs: + ... print(job.progress) """ data = self._request("GET", f"/printers/{printer_uuid}/jobs") if isinstance(data, dict) and "jobs" in data: @@ -220,6 +280,9 @@ def send_command(self, printer_uuid: str, command: str, kwargs: dict | None = No Returns: True if successful. + + Usage Example: + >>> success = client.send_command("c0ffee-1234", "PAUSE_PRINT") """ payload = {"command": command} if kwargs: @@ -237,6 +300,11 @@ def get_snapshot(self, camera_id: str) -> bytes: Returns: The binary image data. + + Usage Example: + >>> image_data = client.get_snapshot("camera_1") + >>> with open("snap.jpg", "wb") as f: + ... f.write(image_data) """ # Raw response for binary data response = self._request("GET", f"/cameras/{camera_id}/snapshots/last", raw=True) @@ -250,7 +318,9 @@ def trigger_snapshot(self, camera_token: str) -> bool: Returns: True if triggered. + + Usage Example: + >>> client.trigger_snapshot("token_abc_123") """ self._request("POST", f"/cameras/{camera_token}/snapshots") return True - diff --git a/tests/test_client_improvements.py b/tests/test_client_improvements.py new file mode 100644 index 0000000..cd48f86 --- /dev/null +++ b/tests/test_client_improvements.py @@ -0,0 +1,86 @@ +from unittest.mock import patch + +import pytest +import responses +from requests.exceptions import ReadTimeout + +from prusa_connect.client import DEFAULT_TIMEOUT, PrusaConnectClient +from prusa_connect.exceptions import PrusaApiError, PrusaNetworkError + + +class MockCredentials: + def before_request(self, headers: dict[str, str]) -> None: + headers["Authorization"] = "Bearer mock_token" + + +@pytest.fixture +def client(): + return PrusaConnectClient(credentials=MockCredentials()) + + +def test_timeout_arg_passed_to_session(): + """Verify session.request receives timeout.""" + creds = MockCredentials() + client = PrusaConnectClient(creds) + + with patch.object(client._session, "request") as mock_request: + mock_request.return_value.status_code = 200 + mock_request.return_value.json.return_value = {"printers": []} + mock_request.return_value.headers = {} + mock_request.return_value.content = b"{}" + + client.get_printers() + + mock_request.assert_called_with( + "GET", + "https://connect.prusa3d.com/app/printers", + timeout=DEFAULT_TIMEOUT + ) + + +@responses.activate +def test_json_error_parsing(client): + """Verify that JSON error messages are parsed into PrusaApiError.""" + error_message = "Printer is busy doing something else." + responses.add( + responses.GET, + "https://connect.prusa3d.com/app/printers", + json={"error": error_message}, + status=400, + ) + + with pytest.raises(PrusaApiError) as exc_info: + client.get_printers() + + assert error_message in str(exc_info.value) + assert exc_info.value.status_code == 400 + + +@responses.activate +def test_json_error_parsing_nested_message(client): + """Verify that JSON error messages with 'message' key are parsed.""" + error_message = "Something went wrong." + responses.add( + responses.GET, + "https://connect.prusa3d.com/app/printers", + json={"message": error_message}, + status=500, + ) + + with pytest.raises(PrusaApiError) as exc_info: + client.get_printers() + + assert error_message in str(exc_info.value) + + +@responses.activate +def test_network_timeout_wraps_exception(client): + """Verify that requests.Timeout is wrapped in PrusaNetworkError.""" + responses.add( + responses.GET, + "https://connect.prusa3d.com/app/printers", + body=ReadTimeout("Connection timed out"), + ) + + with pytest.raises(PrusaNetworkError): + client.get_printers() diff --git a/uv.lock b/uv.lock index a24ccc3..104e05b 100644 --- a/uv.lock +++ b/uv.lock @@ -1,6 +1,6 @@ version = 1 revision = 3 -requires-python = ">=3.14" +requires-python = ">=3.12" [[package]] name = "annotated-types"