From 4b148f94b3fbb467e7b1f6be0e717271ceadac33 Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 20 Jun 2026 17:14:54 -0300 Subject: [PATCH 1/5] feat(cli): add auth command to verify GP_ACCESS_TOKEN Expose check_auth() on API and client layers with GoProAuthStatus, and add gopro-api auth with Rich, JSON, and TSV output plus exit codes. Co-authored-by: Cursor --- gopro_api/api/async_gopro.py | 104 +++++++++++++++++- gopro_api/api/gopro.py | 105 ++++++++++++++++++- gopro_api/api/models.py | 20 ++++ gopro_api/cli/__init__.py | 10 +- gopro_api/cli/auth.py | 197 +++++++++++++++++++++++++++++++++++ gopro_api/client.py | 23 ++++ gopro_api/config.py | 18 +++- 7 files changed, 473 insertions(+), 4 deletions(-) create mode 100644 gopro_api/cli/auth.py diff --git a/gopro_api/api/async_gopro.py b/gopro_api/api/async_gopro.py index a1bf1cf..e4f6278 100644 --- a/gopro_api/api/async_gopro.py +++ b/gopro_api/api/async_gopro.py @@ -2,8 +2,9 @@ import aiohttp -from gopro_api.config import get_settings +from gopro_api.config import get_settings, get_token_info from gopro_api.api.models import ( + GoProAuthStatus, GoProMediaSearchParams, GoProMediaDownloadResponse, GoProMediaSearchResponse, @@ -26,6 +27,7 @@ def __init__(self, access_token: str | None = None, timeout: float = 10.0) -> No :attr:`~gopro_api.config.Settings.gp_access_token` from settings. timeout: Total client timeout in seconds for ``aiohttp``. """ + self._explicit_token = access_token is not None self.access_token = access_token or get_settings().gp_access_token self._timeout = aiohttp.ClientTimeout(total=timeout) self._session: aiohttp.ClientSession | None = None @@ -149,3 +151,103 @@ async def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchRespon response.raise_for_status() body = await response.text() return GoProMediaSearchResponse.model_validate_json(body) + + def _token_source(self) -> str | None: + """Return where the active access token was loaded from. + + Returns: + ``"argument"``, ``"environment"``, ``".env"``, or ``None`` when unset. + """ + if not self.access_token: + return None + if self._explicit_token: + return "argument" + _, source = get_token_info() + return source + + def _auth_status_without_request(self) -> GoProAuthStatus: + """Build a status result when no token is configured. + + Returns: + Status indicating the token is missing. + """ + return GoProAuthStatus( + token_configured=False, + token_source=None, + authenticated=None, + http_status=None, + message="GP_ACCESS_TOKEN is not set.", + ) + + def _auth_status_from_http(self, *, status_code: int, source: str) -> GoProAuthStatus: + """Build a status result from an HTTP verification response. + + Args: + status_code: HTTP status returned by ``GET /media/search``. + source: Token source label from :meth:`_token_source`. + + Returns: + Parsed authentication status for the caller. + """ + if status_code == 200: + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=True, + http_status=status_code, + message="Access token is valid.", + ) + if status_code == 401: + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=status_code, + message="Access token was rejected (expired or invalid).", + ) + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=status_code, + message=f"Unexpected HTTP status {status_code}.", + ) + + async def check_auth(self) -> GoProAuthStatus: + """Verify that the configured access token is accepted by the API. + + Performs a lightweight ``GET /media/search`` request with ``per_page=1``. + + Returns: + Structured authentication status; never raises for HTTP failures. + + Raises: + RuntimeError: If used outside ``async with AsyncGoProAPI() as api``. + """ + if not self.access_token: + return self._auth_status_without_request() + + headers = self.get_headers( + "application/vnd.gopro.jk.media.search+json; version=2.0.0", + ) + params = GoProMediaSearchParams(per_page=1, page=1) + session = self._session_or_raise() + source = self._token_source() + try: + async with session.get( + "/media/search", + headers=headers, + params=params.model_dump(), + ) as response: + return self._auth_status_from_http( + status_code=response.status, + source=source or "argument", + ) + except aiohttp.ClientError as exc: + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=None, + message=f"Request failed: {exc}", + ) diff --git a/gopro_api/api/gopro.py b/gopro_api/api/gopro.py index 5f2eb1e..2ef1380 100644 --- a/gopro_api/api/gopro.py +++ b/gopro_api/api/gopro.py @@ -2,8 +2,9 @@ import requests -from gopro_api.config import get_settings +from gopro_api.config import get_settings, get_token_info from gopro_api.api.models import ( + GoProAuthStatus, GoProMediaDownloadResponse, GoProMediaSearchParams, GoProMediaSearchResponse, @@ -26,6 +27,7 @@ def __init__(self, access_token: str | None = None, timeout: float = 10.0) -> No :attr:`~gopro_api.config.Settings.gp_access_token` from settings. timeout: Per-request timeout in seconds passed to ``requests``. """ + self._explicit_token = access_token is not None self.access_token = access_token or get_settings().gp_access_token self._timeout = timeout self._session: requests.Session | None = None @@ -142,3 +144,104 @@ def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchResponse: ) response.raise_for_status() return GoProMediaSearchResponse.model_validate_json(response.text) + + def _token_source(self) -> str | None: + """Return where the active access token was loaded from. + + Returns: + ``"argument"``, ``"environment"``, ``".env"``, or ``None`` when unset. + """ + if not self.access_token: + return None + if self._explicit_token: + return "argument" + _, source = get_token_info() + return source + + def _auth_status_without_request(self) -> GoProAuthStatus: + """Build a status result when no token is configured. + + Returns: + Status indicating the token is missing. + """ + return GoProAuthStatus( + token_configured=False, + token_source=None, + authenticated=None, + http_status=None, + message="GP_ACCESS_TOKEN is not set.", + ) + + def _auth_status_from_http(self, *, status_code: int, source: str) -> GoProAuthStatus: + """Build a status result from an HTTP verification response. + + Args: + status_code: HTTP status returned by ``GET /media/search``. + source: Token source label from :meth:`_token_source`. + + Returns: + Parsed authentication status for the caller. + """ + if status_code == 200: + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=True, + http_status=status_code, + message="Access token is valid.", + ) + if status_code == 401: + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=status_code, + message="Access token was rejected (expired or invalid).", + ) + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=status_code, + message=f"Unexpected HTTP status {status_code}.", + ) + + def check_auth(self) -> GoProAuthStatus: + """Verify that the configured access token is accepted by the API. + + Performs a lightweight ``GET /media/search`` request with ``per_page=1``. + + Returns: + Structured authentication status; never raises for HTTP failures. + + Raises: + RuntimeError: If used outside ``with GoProAPI() as api``. + """ + if not self.access_token: + return self._auth_status_without_request() + + headers = self.get_headers( + "application/vnd.gopro.jk.media.search+json; version=2.0.0", + ) + params = GoProMediaSearchParams(per_page=1, page=1) + session = self._session_or_raise() + source = self._token_source() + try: + response = session.get( + f"{self.base_url}/media/search", + headers=headers, + params=params.model_dump(), + timeout=self._timeout, + ) + except requests.RequestException as exc: + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=None, + message=f"Request failed: {exc}", + ) + return self._auth_status_from_http( + status_code=response.status_code, + source=source or "argument", + ) diff --git a/gopro_api/api/models.py b/gopro_api/api/models.py index 5cd8ee3..ca3e688 100644 --- a/gopro_api/api/models.py +++ b/gopro_api/api/models.py @@ -289,3 +289,23 @@ class GoProMediaDownloadResponse(BaseModel): filename: str embedded: GoProMediaDownloadEmbedded = Field(alias="_embedded") + + +class GoProAuthStatus(BaseModel): + """Result of verifying ``GP_ACCESS_TOKEN`` against the GoPro cloud API. + + Attributes: + token_configured: Whether a non-empty access token is available. + token_source: Where the token came from (``environment``, ``.env``, + ``argument``), or ``None`` when not configured. + authenticated: ``True`` when the API accepted the token, ``False`` when + rejected or the request failed, ``None`` when no token was configured. + http_status: HTTP status from the verification request, if one was made. + message: Human-readable summary suitable for CLI or logs. + """ + + token_configured: bool + token_source: str | None = None + authenticated: bool | None = None + http_status: int | None = None + message: str diff --git a/gopro_api/cli/__init__.py b/gopro_api/cli/__init__.py index 2abf5bc..06b64ce 100644 --- a/gopro_api/cli/__init__.py +++ b/gopro_api/cli/__init__.py @@ -7,5 +7,13 @@ from .search import search_command from .info import info_command from .pull import pull_command +from .auth import auth_command -__all__ = ["app", "main", "search_command", "info_command", "pull_command"] +__all__ = [ + "app", + "main", + "search_command", + "info_command", + "pull_command", + "auth_command", +] diff --git a/gopro_api/cli/auth.py b/gopro_api/cli/auth.py new file mode 100644 index 0000000..049fd23 --- /dev/null +++ b/gopro_api/cli/auth.py @@ -0,0 +1,197 @@ +"""Auth command: AuthPrinter class, async runner, and Typer callback.""" + +from __future__ import annotations + +import asyncio +import json +import sys + +import typer +from rich.console import Console +from rich.panel import Panel +from rich.status import Status +from rich.table import Table +from rich import box + +from gopro_api.api.models import GoProAuthStatus +from gopro_api.client import AsyncGoProClient + +from .app import app +from ._common import _yes_no + + +class AuthPrinter: + """Handles Rich, TSV, and JSON rendering for the auth command.""" + + def __init__(self, console: Console | None = None) -> None: + """Initialize with an optional Rich console. + + Args: + console: Console used for Rich output; a default soft-wrap console is + created when ``None``. + """ + self._console = console or Console(soft_wrap=True) + self._active_status: Status | None = None + + def start_stage(self) -> None: + """Show a spinner while the API verification request runs.""" + self._active_status = self._console.status( + "⏳ [bold cyan]Verifying access token…[/bold cyan]", + ) + self._active_status.__enter__() + + def stop_stage(self) -> None: + """Stop the active spinner, if any.""" + if self._active_status is not None: + self._active_status.__exit__(None, None, None) + self._active_status = None + + def _authenticated_label(self, status: GoProAuthStatus) -> str: + """Format the authenticated field for display. + + Args: + status: Authentication status from the API client. + + Returns: + Human-readable yes/no/not checked label. + """ + if status.authenticated is None: + return "not checked" + return _yes_no(status.authenticated) + + def print_rich(self, status: GoProAuthStatus) -> None: + """Print authentication status as a Rich panel. + + Args: + status: Authentication status from the API client. + """ + table = Table(show_header=False, box=box.SIMPLE) + table.add_column("field", style="bold") + table.add_column("value") + table.add_row("token configured", _yes_no(status.token_configured)) + table.add_row( + "token source", + status.token_source or "—", + ) + table.add_row("authenticated", self._authenticated_label(status)) + if status.http_status is not None: + table.add_row("http status", str(status.http_status)) + table.add_row("message", status.message) + + border = "green" if status.authenticated else "red" + if not status.token_configured: + border = "yellow" + elif status.authenticated is None: + border = "yellow" + + self._console.print( + Panel( + table, + title="[bold blue]Authentication[/bold blue]", + border_style=border, + ), + ) + + def print_tsv(self, status: GoProAuthStatus) -> None: + """Print authentication status as tab-separated key/value rows. + + Args: + status: Authentication status from the API client. + """ + rows = { + "token_configured": _yes_no(status.token_configured), + "token_source": status.token_source or "", + "authenticated": self._authenticated_label(status), + "http_status": "" if status.http_status is None else str(status.http_status), + "message": status.message, + } + console = Console(soft_wrap=True, highlight=False, markup=False) + console.print("field\tvalue") + for key, value in rows.items(): + console.print(f"{key}\t{value}") + + def print_json(self, status: GoProAuthStatus) -> None: + """Print authentication status as JSON on stdout. + + Args: + status: Authentication status from the API client. + """ + sys.stdout.write(json.dumps(status.model_dump(mode="json"), indent=2)) + sys.stdout.write("\n") + + def exit_code(self, status: GoProAuthStatus) -> int: + """Map authentication status to a process exit code. + + Args: + status: Authentication status from the API client. + + Returns: + ``0`` when authenticated, ``2`` when the token is missing, ``1`` otherwise. + """ + if not status.token_configured: + return 2 + if status.authenticated: + return 0 + return 1 + + +async def _run_auth( + *, + timeout: float, + json_out: bool, + tsv: bool, +) -> None: + """Verify ``GP_ACCESS_TOKEN`` and render the result. + + Args: + timeout: HTTP timeout in seconds passed to ``AsyncGoProClient``. + json_out: When ``True``, emit JSON instead of a Rich panel. + tsv: When ``True``, emit tab-separated values instead of a Rich panel. + """ + printer = AuthPrinter() + if not json_out and not tsv: + printer.start_stage() + try: + async with AsyncGoProClient(timeout=timeout) as client: + status = await client.check_auth() + finally: + printer.stop_stage() + + if json_out: + printer.print_json(status) + elif tsv: + printer.print_tsv(status) + else: + printer.print_rich(status) + + raise typer.Exit(printer.exit_code(status)) + + +@app.command( + "auth", + help=( + "Verify GP_ACCESS_TOKEN configuration and authentication " + "(Rich panel by default; --tsv or --json for scripting)" + ), +) +def auth_command( + ctx: typer.Context, + tsv: bool = typer.Option( + False, + "--tsv", + help="Print tab-separated values for scripting", + ), + json_out: bool = typer.Option( + False, + "--json", + help="Print structured JSON", + ), +) -> None: + """Check whether the GoPro access token is configured and accepted by the API.""" + asyncio.run( + _run_auth( + timeout=ctx.obj["timeout"], + json_out=json_out, + tsv=tsv, + ), + ) diff --git a/gopro_api/client.py b/gopro_api/client.py index 67bda61..b069b01 100644 --- a/gopro_api/client.py +++ b/gopro_api/client.py @@ -14,6 +14,7 @@ from gopro_api.api.gopro import GoProAPI from gopro_api.api.models import ( CapturedRange, + GoProAuthStatus, GoProMediaDownloadResponse, GoProMediaSearchItem, GoProMediaSearchParams, @@ -115,6 +116,17 @@ def download(self, media_id: str) -> GoProMediaDownloadResponse: """ return self._api.download(media_id) + def check_auth(self) -> GoProAuthStatus: + """Verify that the configured access token is accepted by the API. + + Returns: + Structured authentication status from the underlying API client. + + Raises: + RuntimeError: If used outside ``with GoProClient()``. + """ + return self._api.check_auth() + # ------------------------------------------------------------------ # High-level helpers # ------------------------------------------------------------------ @@ -318,6 +330,17 @@ async def download(self, media_id: str) -> GoProMediaDownloadResponse: """ return await self._api.download(media_id) + async def check_auth(self) -> GoProAuthStatus: + """Verify that the configured access token is accepted by the API. + + Returns: + Structured authentication status from the underlying API client. + + Raises: + RuntimeError: If used outside ``async with AsyncGoProClient()``. + """ + return await self._api.check_auth() + # ------------------------------------------------------------------ # High-level helpers # ------------------------------------------------------------------ diff --git a/gopro_api/config.py b/gopro_api/config.py index 73f97c8..07f3e9e 100644 --- a/gopro_api/config.py +++ b/gopro_api/config.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os from functools import lru_cache from pydantic_settings import BaseSettings, SettingsConfigDict @@ -40,4 +41,19 @@ def get_settings() -> Settings: return Settings() -__all__ = ["Settings", "get_settings"] +def get_token_info() -> tuple[bool, str | None]: + """Report whether ``GP_ACCESS_TOKEN`` is available and where it came from. + + Returns: + A pair ``(configured, source)`` where ``source`` is ``"environment"``, + ``".env"``, or ``None`` when not configured. + """ + token = get_settings().gp_access_token + if not token: + return False, None + if os.environ.get("GP_ACCESS_TOKEN"): + return True, "environment" + return True, ".env" + + +__all__ = ["Settings", "get_settings", "get_token_info"] From b6c87443908a259f62828ddbdc9aafadf1017b16 Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 20 Jun 2026 17:15:07 -0300 Subject: [PATCH 2/5] docs: document auth command and check_auth API Update CLI, configuration, architecture, and API reference pages for gopro-api auth and the new GoProAuthStatus model. Co-authored-by: Cursor --- README.md | 6 +++++- docs/ARCHITECTURE.md | 27 +++++++++++++++++++++++++-- docs/api/cli.md | 4 ++++ docs/api/models.md | 4 ++++ docs/cli.md | 37 +++++++++++++++++++++++++++++++++++++ docs/configuration.md | 7 +++++++ docs/getting-started.md | 6 ++++++ docs/index.md | 2 +- 8 files changed, 89 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 868066a..8fc9169 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ This project is not affiliated with or endorsed by GoPro. - **`GoProAPI`** — synchronous client (`requests`), `with` context manager - **`AsyncGoProAPI`** — async client (`aiohttp`), `async with` context manager - **Pydantic** request/response types in `gopro_api.api.models` -- **CLI** — `gopro-api search`, `gopro-api info`, `gopro-api pull` +- **CLI** — `gopro-api search`, `gopro-api info`, `gopro-api pull`, `gopro-api auth` - **`GP_ACCESS_TOKEN`** from environment / `.env` (browser cookie value) ## Requirements @@ -62,6 +62,8 @@ gopro-api info MEDIA_ID --json gopro-api pull MEDIA_ID ./downloads gopro-api pull MEDIA_ID ./downloads --height 1080 gopro-api pull MEDIA_ID ./downloads --width 1920 --height 1080 +gopro-api auth +gopro-api auth --json ``` | Command | Purpose | @@ -69,6 +71,7 @@ gopro-api pull MEDIA_ID ./downloads --width 1920 --height 1080 | **`search`** | List media in a capture range. Default: a **`# _pages`** summary line, a tab-separated header (`id`, `type`, `captured_at`, `filename`, …; not `gopro_user_id` / `source_gumi` / `source_mgumi`), then one row per item (other API fields in an **`extra`** JSON column). **`--json`**: full API-shaped response; with **`--all-pages`**, a JSON array of every page. | | **`info`** | Show download metadata for one media id (filename + file lines with size and URL), or **`--json`** for the full payload. | | **`pull`** | Download asset(s) for a media id into **`destination`** (directory; created if missing). Videos (`.mp4` extension, case-insensitive): one **`variations`** entry — **tallest** by default, or closest to **`--height`** / **`--width`** (sum of squared pixel deltas; ties broken by larger resolution). Photos: uses **`files`** (one request per file). | +| **`auth`** | Verify that **`GP_ACCESS_TOKEN`** is configured and accepted by the API. Default: Rich panel; **`--json`** / **`--tsv`** for scripting. Exit code **`0`** when authenticated, **`2`** when the token is missing, **`1`** otherwise. | Global **`--timeout`** (seconds, default **`60`**) applies to API calls and to **`pull`** CDN downloads (`requests.get`). @@ -79,6 +82,7 @@ python -m gopro_api.cli search --start 2026-03-01 --end 2026-03-02 python -m gopro_api.cli info MEDIA_ID python -m gopro_api.cli pull MEDIA_ID ./out python -m gopro_api.cli pull MEDIA_ID ./out --height 720 +python -m gopro_api.cli auth ``` ## Configuration diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d55bbf6..312b134 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -30,7 +30,7 @@ flowchart TD | `gopro_api/config.py` | pydantic-settings `Settings` (`GP_ACCESS_TOKEN`) | | `gopro_api/exceptions.py` | Custom exception hierarchy | | `gopro_api/utils.py` | Shared helpers (resolution scoring, etc.) | -| `gopro_api/cli/` | Typer application — `search`, `info`, `pull` commands | +| `gopro_api/cli/` | Typer application — `search`, `info`, `pull`, `auth` commands | ## Layers @@ -65,7 +65,7 @@ async with AsyncGoProAPI() as api: # opens an aiohttp.ClientSession All request parameters and API responses are typed with [Pydantic v2](https://docs.pydantic.dev/) models: - **Requests** — `GoProMediaSearchParams`, `CapturedRange`. -- **Responses** — `GoProMediaSearchResponse`, `GoProMediaDownloadResponse`, and their `_embedded` / `_pages` children (with field aliases for the GoPro API's underscore-prefixed keys). +- **Responses** — `GoProMediaSearchResponse`, `GoProMediaDownloadResponse`, `GoProAuthStatus`, and their `_embedded` / `_pages` children (with field aliases for the GoPro API's underscore-prefixed keys). List fields in request models are serialised to comma-separated strings automatically when calling `model_dump()`. @@ -78,6 +78,8 @@ A single `Settings` class (pydantic-settings) reads `GP_ACCESS_TOKEN` from: Clients accept an explicit `access_token` parameter that overrides `Settings`. +`get_token_info()` reports whether a token is configured and whether it came from the environment or a `.env` file. + ### 5. CLI (`gopro_api.cli`) The CLI is built with [Typer](https://typer.tiangolo.com/) and [Rich](https://github.com/Textualize/rich): @@ -86,6 +88,7 @@ The CLI is built with [Typer](https://typer.tiangolo.com/) and [Rich](https://gi - `gopro_api/cli/search.py` — `search` command + `SearchPrinter`. - `gopro_api/cli/info.py` — `info` command + `InfoPrinter`. - `gopro_api/cli/pull.py` — `pull` command + `PullPrinter`. +- `gopro_api/cli/auth.py` — `auth` command + `AuthPrinter`. - `gopro_api/cli/_common.py` — shared helpers. Each command delegates to `GoProClient`/`AsyncGoProClient` and passes the result to a dedicated `*Printer` class for Rich-formatted output. @@ -110,6 +113,26 @@ sequenceDiagram CLI->>User: Rich table or raw JSON (stdout) ``` +## Data flow — `gopro-api auth` + +```mermaid +sequenceDiagram + actor User + participant CLI + participant AsyncGoProClient + participant AsyncGoProAPI + participant GoproCom as api.gopro.com + + User->>CLI: gopro-api auth + CLI->>AsyncGoProClient: check_auth() + AsyncGoProClient->>AsyncGoProAPI: check_auth() + AsyncGoProAPI->>GoproCom: GET /media/search?per_page=1 + GoproCom-->>AsyncGoProAPI: HTTP 200 or 401 + AsyncGoProAPI-->>AsyncGoProClient: GoProAuthStatus + AsyncGoProClient-->>CLI: GoProAuthStatus + CLI->>User: Rich panel or JSON (stdout) +``` + ## Error handling ```mermaid diff --git a/docs/api/cli.md b/docs/api/cli.md index 3fad183..e268eea 100644 --- a/docs/api/cli.md +++ b/docs/api/cli.md @@ -20,6 +20,8 @@ Built with [Typer](https://typer.tiangolo.com/); the Typer application is expose ::: gopro_api.cli.pull_command +::: gopro_api.cli.auth_command + ## Printers ::: gopro_api.cli.search.SearchPrinter @@ -27,3 +29,5 @@ Built with [Typer](https://typer.tiangolo.com/); the Typer application is expose ::: gopro_api.cli.info.InfoPrinter ::: gopro_api.cli.pull.PullPrinter + +::: gopro_api.cli.auth.AuthPrinter diff --git a/docs/api/models.md b/docs/api/models.md index 83e3881..ece12a0 100644 --- a/docs/api/models.md +++ b/docs/api/models.md @@ -27,3 +27,7 @@ Pydantic models for GoPro cloud media search and download responses. ::: gopro_api.api.models.GoProMediaDownloadFile ::: gopro_api.api.models.GoProMediaDownloadSidecarFile + +## Auth + +::: gopro_api.api.models.GoProAuthStatus diff --git a/docs/cli.md b/docs/cli.md index e44861f..fd6a363 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -117,6 +117,42 @@ gopro-api pull MEDIA_ID ./downloads --height 1080 gopro-api pull MEDIA_ID ./downloads --width 1920 --height 1080 ``` +--- + +### `auth` + +Verify that `GP_ACCESS_TOKEN` is configured and accepted by the GoPro cloud API. + +```bash +gopro-api auth +``` + +**Options** + +| Option | Description | +|--------|-------------| +| `--json` | Print structured JSON instead of the Rich panel. | +| `--tsv` | Print tab-separated key/value rows for scripting. | + +**Default output** — a Rich panel showing whether the token is configured, where it came from (`environment`, `.env`, or an explicit client argument), whether the API accepted it, and a short message. + +**Exit codes** + +| Code | Meaning | +|------|---------| +| `0` | Token configured and authenticated. | +| `1` | Token present but rejected or the verification request failed. | +| `2` | `GP_ACCESS_TOKEN` is not set. | + +```bash +# Rich panel (default) +gopro-api auth + +# Scripting +gopro-api auth --json +gopro-api auth --tsv +``` + ## Running without an installed entry point ```bash @@ -124,6 +160,7 @@ python -m gopro_api.cli search --start 2026-03-01 --end 2026-03-02 python -m gopro_api.cli info MEDIA_ID python -m gopro_api.cli pull MEDIA_ID ./out python -m gopro_api.cli pull MEDIA_ID ./out --height 720 +python -m gopro_api.cli auth ``` ## API reference diff --git a/docs/configuration.md b/docs/configuration.md index 550e584..d5ad1cb 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -62,6 +62,13 @@ Sign in to [gopro.com](https://gopro.com) or [quik.gopro.com](https://quik.gopro Tokens expire. If you receive a **401 Unauthorized** error, return to your browser and copy a fresh token value. +Check whether your current token is still valid: + +```bash +gopro-api auth +gopro-api auth --json +``` + ## API reference See [`gopro_api.config`](api/utils.md) for the `Settings` class that loads these values. diff --git a/docs/getting-started.md b/docs/getting-started.md index dd3dd69..766cce0 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -43,6 +43,12 @@ GP_ACCESS_TOKEN=your_token_here The library loads it automatically via **pydantic-settings**. See [Configuration](configuration.md) for how to retrieve the token from your browser. +Verify the token before searching or downloading: + +```bash +gopro-api auth +``` + ## Quick start — CLI After installing, `gopro-api` is available on your `PATH`: diff --git a/docs/index.md b/docs/index.md index 5186f28..a20de4d 100644 --- a/docs/index.md +++ b/docs/index.md @@ -18,7 +18,7 @@ --- - Full reference for `gopro-api search`, `info`, and `pull` with all flags and examples. + Full reference for `gopro-api search`, `info`, `pull`, and `auth` with all flags and examples. [:octicons-arrow-right-24: CLI](cli.md) From 0b0875d8160edb1e41dd211bed9fab373ecea9b7 Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 20 Jun 2026 17:15:07 -0300 Subject: [PATCH 3/5] chore(release): 0.0.10 Co-authored-by: Cursor --- pyproject.toml | 2 +- uv.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 4901814..0cd74de 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "gopro-api" -version = "0.0.9" +version = "0.0.10" description = "Unofficial Python client for the GoPro cloud API (api.gopro.com): sync and async clients, Pydantic models, and a CLI." readme = "README.md" license = { file = "LICENSE" } diff --git a/uv.lock b/uv.lock index 6de19b6..dfaf23b 100644 --- a/uv.lock +++ b/uv.lock @@ -494,7 +494,7 @@ wheels = [ [[package]] name = "gopro-api" -version = "0.0.8" +version = "0.0.10" source = { editable = "." } dependencies = [ { name = "aiohttp" }, From 646612c262b065f6b44ea8ab42f94ed9dd36ab6c Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 20 Jun 2026 17:19:24 -0300 Subject: [PATCH 4/5] style: apply black formatting Co-authored-by: Cursor --- gopro_api/api/async_gopro.py | 4 +++- gopro_api/api/gopro.py | 4 +++- gopro_api/cli/auth.py | 4 +++- 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/gopro_api/api/async_gopro.py b/gopro_api/api/async_gopro.py index e4f6278..fe438ce 100644 --- a/gopro_api/api/async_gopro.py +++ b/gopro_api/api/async_gopro.py @@ -179,7 +179,9 @@ def _auth_status_without_request(self) -> GoProAuthStatus: message="GP_ACCESS_TOKEN is not set.", ) - def _auth_status_from_http(self, *, status_code: int, source: str) -> GoProAuthStatus: + def _auth_status_from_http( + self, *, status_code: int, source: str + ) -> GoProAuthStatus: """Build a status result from an HTTP verification response. Args: diff --git a/gopro_api/api/gopro.py b/gopro_api/api/gopro.py index 2ef1380..0be9646 100644 --- a/gopro_api/api/gopro.py +++ b/gopro_api/api/gopro.py @@ -172,7 +172,9 @@ def _auth_status_without_request(self) -> GoProAuthStatus: message="GP_ACCESS_TOKEN is not set.", ) - def _auth_status_from_http(self, *, status_code: int, source: str) -> GoProAuthStatus: + def _auth_status_from_http( + self, *, status_code: int, source: str + ) -> GoProAuthStatus: """Build a status result from an HTTP verification response. Args: diff --git a/gopro_api/cli/auth.py b/gopro_api/cli/auth.py index 049fd23..9037a2b 100644 --- a/gopro_api/cli/auth.py +++ b/gopro_api/cli/auth.py @@ -102,7 +102,9 @@ def print_tsv(self, status: GoProAuthStatus) -> None: "token_configured": _yes_no(status.token_configured), "token_source": status.token_source or "", "authenticated": self._authenticated_label(status), - "http_status": "" if status.http_status is None else str(status.http_status), + "http_status": ( + "" if status.http_status is None else str(status.http_status) + ), "message": status.message, } console = Console(soft_wrap=True, highlight=False, markup=False) From c177609137e25be6045bc3d54b4a7ec842ebfb3c Mon Sep 17 00:00:00 2001 From: Welbert Castro Date: Sat, 20 Jun 2026 17:25:12 -0300 Subject: [PATCH 5/5] refactor(auth): extract shared auth status logic for pylint Move duplicated token verification helpers into AuthStatusResolver and use a context manager for the CLI spinner instead of manual dunder calls. Co-authored-by: Cursor --- gopro_api/api/async_gopro.py | 102 +++------------ gopro_api/api/auth_status.py | 234 +++++++++++++++++++++++++++++++++++ gopro_api/api/gopro.py | 100 +++------------ gopro_api/cli/auth.py | 29 ++--- 4 files changed, 275 insertions(+), 190 deletions(-) create mode 100644 gopro_api/api/auth_status.py diff --git a/gopro_api/api/async_gopro.py b/gopro_api/api/async_gopro.py index fe438ce..a5cb8be 100644 --- a/gopro_api/api/async_gopro.py +++ b/gopro_api/api/async_gopro.py @@ -2,7 +2,8 @@ import aiohttp -from gopro_api.config import get_settings, get_token_info +from gopro_api.config import get_settings +from gopro_api.api.auth_status import AuthCheckRequest, AuthStatusResolver from gopro_api.api.models import ( GoProAuthStatus, GoProMediaSearchParams, @@ -152,69 +153,6 @@ async def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchRespon body = await response.text() return GoProMediaSearchResponse.model_validate_json(body) - def _token_source(self) -> str | None: - """Return where the active access token was loaded from. - - Returns: - ``"argument"``, ``"environment"``, ``".env"``, or ``None`` when unset. - """ - if not self.access_token: - return None - if self._explicit_token: - return "argument" - _, source = get_token_info() - return source - - def _auth_status_without_request(self) -> GoProAuthStatus: - """Build a status result when no token is configured. - - Returns: - Status indicating the token is missing. - """ - return GoProAuthStatus( - token_configured=False, - token_source=None, - authenticated=None, - http_status=None, - message="GP_ACCESS_TOKEN is not set.", - ) - - def _auth_status_from_http( - self, *, status_code: int, source: str - ) -> GoProAuthStatus: - """Build a status result from an HTTP verification response. - - Args: - status_code: HTTP status returned by ``GET /media/search``. - source: Token source label from :meth:`_token_source`. - - Returns: - Parsed authentication status for the caller. - """ - if status_code == 200: - return GoProAuthStatus( - token_configured=True, - token_source=source, - authenticated=True, - http_status=status_code, - message="Access token is valid.", - ) - if status_code == 401: - return GoProAuthStatus( - token_configured=True, - token_source=source, - authenticated=False, - http_status=status_code, - message="Access token was rejected (expired or invalid).", - ) - return GoProAuthStatus( - token_configured=True, - token_source=source, - authenticated=False, - http_status=status_code, - message=f"Unexpected HTTP status {status_code}.", - ) - async def check_auth(self) -> GoProAuthStatus: """Verify that the configured access token is accepted by the API. @@ -226,30 +164,20 @@ async def check_auth(self) -> GoProAuthStatus: Raises: RuntimeError: If used outside ``async with AsyncGoProAPI() as api``. """ - if not self.access_token: - return self._auth_status_without_request() - headers = self.get_headers( - "application/vnd.gopro.jk.media.search+json; version=2.0.0", - ) - params = GoProMediaSearchParams(per_page=1, page=1) - session = self._session_or_raise() - source = self._token_source() - try: + async def perform_request(prepared: AuthCheckRequest) -> int: + session = self._session_or_raise() async with session.get( "/media/search", - headers=headers, - params=params.model_dump(), + headers=prepared.headers, + params=prepared.params, ) as response: - return self._auth_status_from_http( - status_code=response.status, - source=source or "argument", - ) - except aiohttp.ClientError as exc: - return GoProAuthStatus( - token_configured=True, - token_source=source, - authenticated=False, - http_status=None, - message=f"Request failed: {exc}", - ) + return response.status + + return await AuthStatusResolver.verify_async( + access_token=self.access_token, + explicit_token=self._explicit_token, + get_headers=self.get_headers, + perform_request=perform_request, + request_error_type=aiohttp.ClientError, + ) diff --git a/gopro_api/api/auth_status.py b/gopro_api/api/auth_status.py new file mode 100644 index 0000000..070289c --- /dev/null +++ b/gopro_api/api/auth_status.py @@ -0,0 +1,234 @@ +"""Shared helpers for building GoPro authentication status results.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Awaitable, Callable, TypeVar + +from gopro_api.api.models import GoProAuthStatus, GoProMediaSearchParams +from gopro_api.config import get_token_info + +AUTH_SEARCH_ACCEPT = "application/vnd.gopro.jk.media.search+json; version=2.0.0" + +_RequestError = TypeVar("_RequestError", bound=BaseException) + + +@dataclass(frozen=True, slots=True) +class AuthCheckRequest: + """Prepared inputs for a token verification ``GET /media/search`` call.""" + + headers: dict[str, str] + params: dict[str, object] + source: str | None + + +class AuthStatusResolver: + """Build :class:`GoProAuthStatus` values for token verification.""" + + @staticmethod + def token_source(*, access_token: str | None, explicit_token: bool) -> str | None: + """Return where the active access token was loaded from. + + Args: + access_token: Token value used for API requests. + explicit_token: Whether the token was passed to the client constructor. + + Returns: + ``"argument"``, ``"environment"``, ``".env"``, or ``None`` when unset. + """ + if not access_token: + return None + if explicit_token: + return "argument" + _, source = get_token_info() + return source + + @staticmethod + def prepare_check( + *, + access_token: str | None, + explicit_token: bool, + get_headers: Callable[[str], dict[str, str]], + ) -> GoProAuthStatus | AuthCheckRequest: + """Prepare headers and query params for token verification. + + Args: + access_token: Token value used for API requests. + explicit_token: Whether the token was passed to the client constructor. + get_headers: Client callback that builds request headers for an Accept + value. + + Returns: + Missing-token status, or request inputs when verification can proceed. + """ + if not access_token: + return AuthStatusResolver.without_request() + + params = GoProMediaSearchParams(per_page=1, page=1) + return AuthCheckRequest( + headers=get_headers(AUTH_SEARCH_ACCEPT), + params=params.model_dump(), + source=AuthStatusResolver.token_source( + access_token=access_token, + explicit_token=explicit_token, + ), + ) + + @staticmethod + def without_request() -> GoProAuthStatus: + """Build a status result when no token is configured. + + Returns: + Status indicating the token is missing. + """ + return GoProAuthStatus( + token_configured=False, + token_source=None, + authenticated=None, + http_status=None, + message="GP_ACCESS_TOKEN is not set.", + ) + + @staticmethod + def from_http(*, status_code: int, source: str) -> GoProAuthStatus: + """Build a status result from an HTTP verification response. + + Args: + status_code: HTTP status returned by ``GET /media/search``. + source: Token source label from :meth:`token_source`. + + Returns: + Parsed authentication status for the caller. + """ + if status_code == 200: + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=True, + http_status=status_code, + message="Access token is valid.", + ) + if status_code == 401: + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=status_code, + message="Access token was rejected (expired or invalid).", + ) + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=status_code, + message=f"Unexpected HTTP status {status_code}.", + ) + + @staticmethod + def from_request_failure( + *, + source: str | None, + message: str, + ) -> GoProAuthStatus: + """Build a status result when the verification request fails. + + Args: + source: Token source label from :meth:`token_source`. + message: Human-readable failure summary. + + Returns: + Parsed authentication status for the caller. + """ + return GoProAuthStatus( + token_configured=True, + token_source=source, + authenticated=False, + http_status=None, + message=message, + ) + + @staticmethod + def verify_sync( + *, + access_token: str | None, + explicit_token: bool, + get_headers: Callable[[str], dict[str, str]], + perform_request: Callable[[AuthCheckRequest], int], + request_error_type: type[_RequestError], + ) -> GoProAuthStatus: + """Run a synchronous token verification request. + + Args: + access_token: Token value used for API requests. + explicit_token: Whether the token was passed to the client constructor. + get_headers: Client callback that builds request headers for an Accept + value. + perform_request: Callable that performs ``GET /media/search`` and returns + the HTTP status code. + request_error_type: Exception type raised when the HTTP client fails. + + Returns: + Structured authentication status; never raises for HTTP failures. + """ + prepared = AuthStatusResolver.prepare_check( + access_token=access_token, + explicit_token=explicit_token, + get_headers=get_headers, + ) + if isinstance(prepared, GoProAuthStatus): + return prepared + + try: + status_code = perform_request(prepared) + except request_error_type as exc: + return AuthStatusResolver.from_request_failure( + source=prepared.source, + message=f"Request failed: {exc}", + ) + return AuthStatusResolver.from_http( + status_code=status_code, + source=prepared.source or "argument", + ) + + @staticmethod + async def verify_async( + *, + access_token: str | None, + explicit_token: bool, + get_headers: Callable[[str], dict[str, str]], + perform_request: Callable[[AuthCheckRequest], Awaitable[int]], + request_error_type: type[_RequestError], + ) -> GoProAuthStatus: + """Run an asynchronous token verification request. + + Args: + access_token: Token value used for API requests. + explicit_token: Whether the token was passed to the client constructor. + get_headers: Client callback that builds request headers for an Accept + value. + perform_request: Callable that performs ``GET /media/search`` and returns + the HTTP status code. + request_error_type: Exception type raised when the HTTP client fails. + + Returns: + Structured authentication status; never raises for HTTP failures. + """ + prepared = AuthStatusResolver.prepare_check( + access_token=access_token, + explicit_token=explicit_token, + get_headers=get_headers, + ) + if isinstance(prepared, GoProAuthStatus): + return prepared + + try: + status_code = await perform_request(prepared) + except request_error_type as exc: + return AuthStatusResolver.from_request_failure( + source=prepared.source, + message=f"Request failed: {exc}", + ) + return AuthStatusResolver.from_http( + status_code=status_code, + source=prepared.source or "argument", + ) diff --git a/gopro_api/api/gopro.py b/gopro_api/api/gopro.py index 0be9646..dc7a18b 100644 --- a/gopro_api/api/gopro.py +++ b/gopro_api/api/gopro.py @@ -2,7 +2,8 @@ import requests -from gopro_api.config import get_settings, get_token_info +from gopro_api.config import get_settings +from gopro_api.api.auth_status import AuthCheckRequest, AuthStatusResolver from gopro_api.api.models import ( GoProAuthStatus, GoProMediaDownloadResponse, @@ -145,69 +146,6 @@ def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchResponse: response.raise_for_status() return GoProMediaSearchResponse.model_validate_json(response.text) - def _token_source(self) -> str | None: - """Return where the active access token was loaded from. - - Returns: - ``"argument"``, ``"environment"``, ``".env"``, or ``None`` when unset. - """ - if not self.access_token: - return None - if self._explicit_token: - return "argument" - _, source = get_token_info() - return source - - def _auth_status_without_request(self) -> GoProAuthStatus: - """Build a status result when no token is configured. - - Returns: - Status indicating the token is missing. - """ - return GoProAuthStatus( - token_configured=False, - token_source=None, - authenticated=None, - http_status=None, - message="GP_ACCESS_TOKEN is not set.", - ) - - def _auth_status_from_http( - self, *, status_code: int, source: str - ) -> GoProAuthStatus: - """Build a status result from an HTTP verification response. - - Args: - status_code: HTTP status returned by ``GET /media/search``. - source: Token source label from :meth:`_token_source`. - - Returns: - Parsed authentication status for the caller. - """ - if status_code == 200: - return GoProAuthStatus( - token_configured=True, - token_source=source, - authenticated=True, - http_status=status_code, - message="Access token is valid.", - ) - if status_code == 401: - return GoProAuthStatus( - token_configured=True, - token_source=source, - authenticated=False, - http_status=status_code, - message="Access token was rejected (expired or invalid).", - ) - return GoProAuthStatus( - token_configured=True, - token_source=source, - authenticated=False, - http_status=status_code, - message=f"Unexpected HTTP status {status_code}.", - ) - def check_auth(self) -> GoProAuthStatus: """Verify that the configured access token is accepted by the API. @@ -219,31 +157,21 @@ def check_auth(self) -> GoProAuthStatus: Raises: RuntimeError: If used outside ``with GoProAPI() as api``. """ - if not self.access_token: - return self._auth_status_without_request() - headers = self.get_headers( - "application/vnd.gopro.jk.media.search+json; version=2.0.0", - ) - params = GoProMediaSearchParams(per_page=1, page=1) - session = self._session_or_raise() - source = self._token_source() - try: + def perform_request(prepared: AuthCheckRequest) -> int: + session = self._session_or_raise() response = session.get( f"{self.base_url}/media/search", - headers=headers, - params=params.model_dump(), + headers=prepared.headers, + params=prepared.params, timeout=self._timeout, ) - except requests.RequestException as exc: - return GoProAuthStatus( - token_configured=True, - token_source=source, - authenticated=False, - http_status=None, - message=f"Request failed: {exc}", - ) - return self._auth_status_from_http( - status_code=response.status_code, - source=source or "argument", + return response.status_code + + return AuthStatusResolver.verify_sync( + access_token=self.access_token, + explicit_token=self._explicit_token, + get_headers=self.get_headers, + perform_request=perform_request, + request_error_type=requests.RequestException, ) diff --git a/gopro_api/cli/auth.py b/gopro_api/cli/auth.py index 9037a2b..2dde0fc 100644 --- a/gopro_api/cli/auth.py +++ b/gopro_api/cli/auth.py @@ -5,11 +5,12 @@ import asyncio import json import sys +from collections.abc import Iterator +from contextlib import contextmanager import typer from rich.console import Console from rich.panel import Panel -from rich.status import Status from rich.table import Table from rich import box @@ -31,20 +32,14 @@ def __init__(self, console: Console | None = None) -> None: created when ``None``. """ self._console = console or Console(soft_wrap=True) - self._active_status: Status | None = None - def start_stage(self) -> None: + @contextmanager + def verifying_status(self) -> Iterator[None]: """Show a spinner while the API verification request runs.""" - self._active_status = self._console.status( + with self._console.status( "⏳ [bold cyan]Verifying access token…[/bold cyan]", - ) - self._active_status.__enter__() - - def stop_stage(self) -> None: - """Stop the active spinner, if any.""" - if self._active_status is not None: - self._active_status.__exit__(None, None, None) - self._active_status = None + ): + yield def _authenticated_label(self, status: GoProAuthStatus) -> str: """Format the authenticated field for display. @@ -151,13 +146,13 @@ async def _run_auth( tsv: When ``True``, emit tab-separated values instead of a Rich panel. """ printer = AuthPrinter() - if not json_out and not tsv: - printer.start_stage() - try: + if json_out or tsv: async with AsyncGoProClient(timeout=timeout) as client: status = await client.check_auth() - finally: - printer.stop_stage() + else: + with printer.verifying_status(): + async with AsyncGoProClient(timeout=timeout) as client: + status = await client.check_auth() if json_out: printer.print_json(status)