Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion gopro_api/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,9 @@
"""Unofficial Python client for the GoPro cloud API (api.gopro.com)."""
"""Unofficial Python client for the GoPro cloud API (``api.gopro.com``).

Exports sync/async low-level API classes, high-level clients, Pydantic models,
the ``NoVariationsError`` exception, and the ``gopro-api`` CLI entrypoint via
``python -m`` / console script.
"""

from gopro_api.api.async_gopro import AsyncGoProAPI
from gopro_api.api.gopro import GoProAPI
Expand Down
6 changes: 5 additions & 1 deletion gopro_api/api/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
"""Sync and async HTTP clients for api.gopro.com."""
"""Low-level sync and async HTTP clients for ``https://api.gopro.com``.

Prefer ``gopro_api.client.GoProClient`` / ``AsyncGoProClient`` for pagination
and downloads unless you need direct control over requests.
"""

from .gopro import GoProAPI
from .async_gopro import AsyncGoProAPI
Expand Down
76 changes: 67 additions & 9 deletions gopro_api/api/async_gopro.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,41 +21,69 @@ class AsyncGoProAPI:
def __init__(self, access_token: str | None = None, timeout: float = 10.0) -> None:
"""Create an async client.

``access_token``: cookie value; defaults to ``GP_ACCESS_TOKEN``.
``timeout``: total HTTP client timeout in seconds.
Args:
access_token: ``gp_access_token`` cookie value; defaults to
``gopro_api.config.GP_ACCESS_TOKEN``.
timeout: Total client timeout in seconds for ``aiohttp``.
"""
self.access_token = access_token or GP_ACCESS_TOKEN
self._timeout = aiohttp.ClientTimeout(total=timeout)
self._session: aiohttp.ClientSession | None = None

@property
def base_url(self) -> str:
"""API origin (``https://api.gopro.com``)."""
"""HTTPS origin for API requests.

Returns:
Always ``https://api.gopro.com``.
"""
return "https://api.gopro.com"

def get_headers(self, accept: str) -> dict[str, str]:
"""Build ``Cookie`` and ``Accept`` headers for an API call."""
"""Build headers for a JSON API request.

Args:
accept: Full ``Accept`` header value (vendor MIME type + version).

Returns:
Mapping with ``Cookie`` (token) and ``Accept``.
"""
return {
"Cookie": "gp_access_token=" + self.access_token,
"Accept": accept,
}

async def __aenter__(self) -> "AsyncGoProAPI":
"""Open an ``aiohttp.ClientSession`` for the ``async with`` body."""
"""Open an ``aiohttp.ClientSession`` for the ``async with`` body.

Returns:
``self`` for use inside the ``async with`` body.
"""
self._session = aiohttp.ClientSession(
base_url=self.base_url,
timeout=self._timeout,
)
return self

async def __aexit__(self, *exc: object) -> None:
"""Close the session."""
"""Close the session and clear internal state.

Args:
*exc: Exception info from the interpreter (ignored).
"""
if self._session is not None:
await self._session.close()
self._session = None

def _session_or_raise(self) -> aiohttp.ClientSession:
"""Return the active session or raise if not inside ``async with``."""
"""Return the active ``aiohttp.ClientSession``.

Returns:
The session opened in ``__aenter__``.

Raises:
RuntimeError: If called before ``__aenter__`` or after ``__aexit__``.
"""
session = self._session
if session is None:
msg = (
Expand All @@ -66,7 +94,22 @@ def _session_or_raise(self) -> aiohttp.ClientSession:
return session

async def download(self, media_id: str) -> GoProMediaDownloadResponse:
"""``GET /media/{media_id}/download`` — metadata and CDN URLs for files."""
"""Return download metadata and CDN URLs for one media item.

Calls ``GET /media/{media_id}/download`` with the GoPro media JSON
vendor MIME type.

Args:
media_id: Cloud library identifier for the media item.

Returns:
Parsed response (filenames, variations, embedded files, CDN URLs).

Raises:
RuntimeError: If used outside ``async with AsyncGoProAPI() as api``.
aiohttp.ClientResponseError: When ``raise_for_status`` fails.
pydantic.ValidationError: If the JSON body does not match the model.
"""
headers = self.get_headers("application/vnd.gopro.jk.media+json; version=2.0.0")
session = self._session_or_raise()
async with session.get(
Expand All @@ -78,7 +121,22 @@ async def download(self, media_id: str) -> GoProMediaDownloadResponse:
return GoProMediaDownloadResponse.model_validate_json(body)

async def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchResponse:
"""``GET /media/search`` using ``params.model_dump()`` as query string."""
"""Search media in the cloud library with structured query parameters.

Calls ``GET /media/search``; ``params.model_dump()`` is sent as the query
string after serialization.

Args:
params: Search filters (capture range, pagination, fields, etc.).

Returns:
Paginated search results and embedded media rows.

Raises:
RuntimeError: If used outside ``async with AsyncGoProAPI() as api``.
aiohttp.ClientResponseError: When ``raise_for_status`` fails.
pydantic.ValidationError: If the JSON body does not match the model.
"""
headers = self.get_headers(
"application/vnd.gopro.jk.media.search+json; version=2.0.0",
)
Expand Down
76 changes: 67 additions & 9 deletions gopro_api/api/gopro.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,45 +21,88 @@ class GoProAPI:
def __init__(self, access_token: str | None = None, timeout: float = 10.0) -> None:
"""Create a sync client.

``access_token``: cookie value; defaults to ``GP_ACCESS_TOKEN``.
``timeout``: per-request timeout in seconds.
Args:
access_token: ``gp_access_token`` cookie value; defaults to
``gopro_api.config.GP_ACCESS_TOKEN``.
timeout: Per-request timeout in seconds passed to ``requests``.
"""
self.access_token = access_token or GP_ACCESS_TOKEN
self._timeout = timeout
self._session: requests.Session | None = None

@property
def base_url(self) -> str:
"""API origin (``https://api.gopro.com``)."""
"""HTTPS origin for API requests.

Returns:
Always ``https://api.gopro.com``.
"""
return "https://api.gopro.com"

def get_headers(self, accept: str) -> dict[str, str]:
"""Build ``Cookie`` and ``Accept`` headers for an API call."""
"""Build headers for a JSON API request.

Args:
accept: Full ``Accept`` header value (vendor MIME type + version).

Returns:
Mapping with ``Cookie`` (token) and ``Accept``.
"""
return {
"Cookie": "gp_access_token=" + self.access_token,
"Accept": accept,
}

def __enter__(self) -> "GoProAPI":
"""Open a ``requests.Session`` for the duration of the ``with`` block."""
"""Open a ``requests.Session`` for the duration of the ``with`` block.

Returns:
``self`` for use inside the ``with`` body.
"""
self._session = requests.Session()
return self

def __exit__(self, *exc: object) -> None:
"""Close the session."""
"""Close the session and clear internal state.

Args:
*exc: Exception info from the interpreter (ignored).
"""
if self._session is not None:
self._session.close()
self._session = None

def _session_or_raise(self) -> requests.Session:
"""Return the active session or raise if used outside a context manager."""
"""Return the active ``requests.Session``.

Returns:
The session opened in ``__enter__``.

Raises:
RuntimeError: If called before ``__enter__`` or after ``__exit__``.
"""
if self._session is None:
msg = "Use GoProAPI as a context manager: with GoProAPI() as api: ..."
raise RuntimeError(msg)
return self._session

def download(self, media_id: str) -> GoProMediaDownloadResponse:
"""``GET /media/{media_id}/download`` — metadata and CDN URLs for files."""
"""Return download metadata and CDN URLs for one media item.

Calls ``GET /media/{media_id}/download`` with the GoPro media JSON
vendor MIME type.

Args:
media_id: Cloud library identifier for the media item.

Returns:
Parsed response (filenames, variations, embedded files, CDN URLs).

Raises:
RuntimeError: If used outside ``with GoProAPI() as api``.
requests.HTTPError: When the response status is not successful.
pydantic.ValidationError: If the JSON body does not match the model.
"""
headers = self.get_headers("application/vnd.gopro.jk.media+json; version=2.0.0")
session = self._session_or_raise()
response = session.get(
Expand All @@ -71,7 +114,22 @@ def download(self, media_id: str) -> GoProMediaDownloadResponse:
return GoProMediaDownloadResponse.model_validate_json(response.text)

def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchResponse:
"""``GET /media/search`` using ``params.model_dump()`` as query string."""
"""Search media in the cloud library with structured query parameters.

Calls ``GET /media/search``; ``params.model_dump()`` is sent as the query
string after serialization.

Args:
params: Search filters (capture range, pagination, fields, etc.).

Returns:
Paginated search results and embedded media rows.

Raises:
RuntimeError: If used outside ``with GoProAPI() as api``.
requests.HTTPError: When the response status is not successful.
pydantic.ValidationError: If the JSON body does not match the model.
"""
headers = self.get_headers(
"application/vnd.gopro.jk.media.search+json; version=2.0.0",
)
Expand Down
Loading
Loading