Skip to content

Client

High-level sync and async clients that wrap the GoPro API with search pagination, asset selection, and file download helpers.

GoProClient

gopro_api.client.GoProClient

High-level sync client for GoPro cloud media.

Wraps GoProAPI via composition and adds search pagination, asset selection, and file download helpers. Use as a context manager; the underlying HTTP session is opened and closed for you.

Source code in gopro_api/client.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
class GoProClient:
    """High-level sync client for GoPro cloud media.

    Wraps ``GoProAPI`` via composition and adds search pagination, asset
    selection, and file download helpers. Use as a context manager; the
    underlying HTTP session is opened and closed for you.
    """

    def __init__(
        self,
        access_token: str | None = None,
        timeout: float = 10.0,
        *,
        page_size: int = 1000,
        max_items: int = 1,
        prefer_height: int | None = None,
        prefer_width: int | None = None,
    ) -> None:
        """Create a sync high-level client.

        Args:
            access_token: ``gp_access_token`` cookie value; defaults to
                ``gopro_api.config.GP_ACCESS_TOKEN``.
            timeout: Per-request HTTP timeout in seconds (API and CDN fetches).
            page_size: Default page size for ``iter_nonempty_search_pages``.
            max_items: Maximum rows returned by ``list_media_items``.
            prefer_height: Preferred video height in pixels for ``get_download_url``.
            prefer_width: Preferred video width in pixels for ``get_download_url``.
        """
        self._api = GoProAPI(access_token=access_token, timeout=timeout)
        self._timeout = timeout
        self.page_size = page_size
        self.max_items = max_items
        self.prefer_height = prefer_height
        self.prefer_width = prefer_width

    def __enter__(self) -> "GoProClient":
        """Enter the underlying ``GoProAPI`` context.

        Returns:
            ``self``.
        """
        self._api.__enter__()
        return self

    def __exit__(self, *exc: object) -> None:
        """Exit the underlying ``GoProAPI`` context."""
        self._api.__exit__(*exc)

    # ------------------------------------------------------------------
    # Low-level proxies (keeps cli.py and other callers unchanged)
    # ------------------------------------------------------------------

    def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchResponse:
        """Run a single media search request.

        Args:
            params: Query parameters for ``GET /media/search``.

        Returns:
            Parsed search response.

        Raises:
            RuntimeError: If used outside ``with GoProClient()``.
            requests.HTTPError: When the HTTP status is not successful.
            pydantic.ValidationError: If the JSON body does not match the model.
        """
        return self._api.search(params)

    def download(self, media_id: str) -> GoProMediaDownloadResponse:
        """Fetch download metadata for one media id.

        Args:
            media_id: Cloud library identifier.

        Returns:
            Parsed download metadata response.

        Raises:
            RuntimeError: If used outside ``with GoProClient()``.
            requests.HTTPError: When the HTTP status is not successful.
            pydantic.ValidationError: If the JSON body does not match the model.
        """
        return self._api.download(media_id)

    # ------------------------------------------------------------------
    # High-level helpers
    # ------------------------------------------------------------------

    def iter_nonempty_search_pages(
        self,
        start_date: datetime,
        end_date: datetime,
        *,
        per_page: int | None = None,
        start_page: int = 1,
    ) -> Iterator[GoProMediaSearchResponse]:
        """Yield search result pages until one returns an empty ``_embedded.media``.

        Args:
            start_date: Capture range start (inclusive semantics per API).
            end_date: Capture range end.
            per_page: Items per page; defaults to ``self.page_size``.
            start_page: First page number to request (1-indexed).

        Yields:
            Each non-empty ``GoProMediaSearchResponse`` page.
        """
        page = start_page
        size = per_page if per_page is not None else self.page_size
        while True:
            params = GoProMediaSearchParams(
                captured_range=CapturedRange(start=start_date, end=end_date),
                page=page,
                per_page=size,
            )
            result = self._api.search(params)
            if not result.embedded.media:
                return
            yield result
            page += 1

    def list_media_items(
        self, start_date: datetime, end_date: datetime
    ) -> list[GoProMediaSearchItem]:
        """Collect media rows across pages up to ``max_items``.

        Args:
            start_date: Capture range start.
            end_date: Capture range end.

        Returns:
            Up to ``self.max_items`` ``GoProMediaSearchItem`` instances.

        Raises:
            RuntimeError: If used outside ``with GoProClient()`` on any underlying
                ``search`` call.
            requests.HTTPError: When any underlying ``search`` HTTP status is not
                successful.
            pydantic.ValidationError: If any underlying ``search`` JSON body does not
                match the model.
        """
        all_media: list[GoProMediaSearchItem] = []
        for page_result in self.iter_nonempty_search_pages(start_date, end_date):
            all_media.extend(page_result.embedded.media)
            if len(all_media) >= self.max_items:
                break
        return all_media[: self.max_items]

    def get_download_url(
        self, media_items: list[GoProMediaSearchItem]
    ) -> dict[str, DownloadAsset]:
        """Resolve download assets for each search row.

        Args:
            media_items: One or more media rows (typically from search).

        Returns:
            Merged mapping of output filename to file or variation metadata.

        Raises:
            NoVariationsError: For video items with no variations.
            RuntimeError: If used outside ``with GoProClient()`` on any underlying
                ``download`` call.
            requests.HTTPError: When any underlying ``download`` HTTP status is not
                successful.
            pydantic.ValidationError: If any underlying ``download`` JSON body does not
                match the model.
        """
        assets: dict[str, DownloadAsset] = {}
        for item in media_items:
            result = self._api.download(item.id)
            assets.update(
                pull_assets_for_response(
                    result,
                    target_height=self.prefer_height,
                    target_width=self.prefer_width,
                )
            )
        return assets

    def download_url_to_path(self, url: str, dest_path: str) -> None:
        """Download a CDN URL to a local path.

        Uses a one-shot ``requests.get`` because CDN hosts differ from
        ``api.gopro.com``.

        Args:
            url: Fully qualified HTTPS URL from download metadata.
            dest_path: Filesystem path for the response body.

        Raises:
            requests.HTTPError: When the HTTP status is not successful.
            OSError: If the destination cannot be written.
        """
        response = requests.get(url, timeout=self._timeout)
        response.raise_for_status()
        dest_dir = os.path.dirname(dest_path)
        if dest_dir:
            os.makedirs(dest_dir, exist_ok=True)
        with open(dest_path, "wb") as out_file:
            out_file.write(response.content)

__init__(access_token=None, timeout=10.0, *, page_size=1000, max_items=1, prefer_height=None, prefer_width=None)

Create a sync high-level client.

Parameters:

Name Type Description Default
access_token str | None

gp_access_token cookie value; defaults to gopro_api.config.GP_ACCESS_TOKEN.

 None
timeout float

Per-request HTTP timeout in seconds (API and CDN fetches).

 10.0
page_size int

Default page size for iter_nonempty_search_pages.

 1000
max_items int

Maximum rows returned by list_media_items.

 1
prefer_height int | None

Preferred video height in pixels for get_download_url.

 None
prefer_width int | None

Preferred video width in pixels for get_download_url.

 None
Source code in gopro_api/client.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    access_token: str | None = None,
    timeout: float = 10.0,
    *,
    page_size: int = 1000,
    max_items: int = 1,
    prefer_height: int | None = None,
    prefer_width: int | None = None,
) -> None:
    """Create a sync high-level client.

    Args:
        access_token: ``gp_access_token`` cookie value; defaults to
            ``gopro_api.config.GP_ACCESS_TOKEN``.
        timeout: Per-request HTTP timeout in seconds (API and CDN fetches).
        page_size: Default page size for ``iter_nonempty_search_pages``.
        max_items: Maximum rows returned by ``list_media_items``.
        prefer_height: Preferred video height in pixels for ``get_download_url``.
        prefer_width: Preferred video width in pixels for ``get_download_url``.
    """
    self._api = GoProAPI(access_token=access_token, timeout=timeout)
    self._timeout = timeout
    self.page_size = page_size
    self.max_items = max_items
    self.prefer_height = prefer_height
    self.prefer_width = prefer_width

__enter__()

Enter the underlying GoProAPI context.

Returns:

Type Description
'GoProClient'

self.
Source code in gopro_api/client.py 
69
70
71
72
73
74
75
76
def __enter__(self) -> "GoProClient":
    """Enter the underlying ``GoProAPI`` context.

    Returns:
        ``self``.
    """
    self._api.__enter__()
    return self

__exit__(*exc)

Exit the underlying GoProAPI context.

Source code in gopro_api/client.py 
78
79
80
def __exit__(self, *exc: object) -> None:
    """Exit the underlying ``GoProAPI`` context."""
    self._api.__exit__(*exc)

search(params)

Run a single media search request.

Parameters:

Name Type Description Default
params GoProMediaSearchParams

Query parameters for GET /media/search.

 required

Returns:

Type Description
GoProMediaSearchResponse

Parsed search response.

Raises:

Type Description
RuntimeError

If used outside with GoProClient().
HTTPError

When the HTTP status is not successful.
ValidationError

If the JSON body does not match the model.
Source code in gopro_api/client.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchResponse:
    """Run a single media search request.

    Args:
        params: Query parameters for ``GET /media/search``.

    Returns:
        Parsed search response.

    Raises:
        RuntimeError: If used outside ``with GoProClient()``.
        requests.HTTPError: When the HTTP status is not successful.
        pydantic.ValidationError: If the JSON body does not match the model.
    """
    return self._api.search(params)

download(media_id)

Fetch download metadata for one media id.

Parameters:

Name Type Description Default
media_id str

Cloud library identifier.

 required

Returns:

Type Description
GoProMediaDownloadResponse

Parsed download metadata response.

Raises:

Type Description
RuntimeError

If used outside with GoProClient().
HTTPError

When the HTTP status is not successful.
ValidationError

If the JSON body does not match the model.
Source code in gopro_api/client.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def download(self, media_id: str) -> GoProMediaDownloadResponse:
    """Fetch download metadata for one media id.

    Args:
        media_id: Cloud library identifier.

    Returns:
        Parsed download metadata response.

    Raises:
        RuntimeError: If used outside ``with GoProClient()``.
        requests.HTTPError: When the HTTP status is not successful.
        pydantic.ValidationError: If the JSON body does not match the model.
    """
    return self._api.download(media_id)

iter_nonempty_search_pages(start_date, end_date, *, per_page=None, start_page=1)

Yield search result pages until one returns an empty _embedded.media.

Parameters:

Name Type Description Default
start_date datetime

Capture range start (inclusive semantics per API).

 required
end_date datetime

Capture range end.

 required
per_page int | None

Items per page; defaults to self.page_size.

 None
start_page int

First page number to request (1-indexed).

 1

Yields:

Type Description
GoProMediaSearchResponse

Each non-empty GoProMediaSearchResponse page.
Source code in gopro_api/client.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def iter_nonempty_search_pages(
    self,
    start_date: datetime,
    end_date: datetime,
    *,
    per_page: int | None = None,
    start_page: int = 1,
) -> Iterator[GoProMediaSearchResponse]:
    """Yield search result pages until one returns an empty ``_embedded.media``.

    Args:
        start_date: Capture range start (inclusive semantics per API).
        end_date: Capture range end.
        per_page: Items per page; defaults to ``self.page_size``.
        start_page: First page number to request (1-indexed).

    Yields:
        Each non-empty ``GoProMediaSearchResponse`` page.
    """
    page = start_page
    size = per_page if per_page is not None else self.page_size
    while True:
        params = GoProMediaSearchParams(
            captured_range=CapturedRange(start=start_date, end=end_date),
            page=page,
            per_page=size,
        )
        result = self._api.search(params)
        if not result.embedded.media:
            return
        yield result
        page += 1

list_media_items(start_date, end_date)

Collect media rows across pages up to max_items.

Parameters:

Name Type Description Default
start_date datetime

Capture range start.

 required
end_date datetime

Capture range end.

 required

Returns:

Type Description
list[GoProMediaSearchItem]

Up to self.max_items GoProMediaSearchItem instances.

Raises:

Type Description
RuntimeError

If used outside with GoProClient() on any underlying search call.
HTTPError

When any underlying search HTTP status is not successful.
ValidationError

If any underlying search JSON body does not match the model.
Source code in gopro_api/client.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def list_media_items(
    self, start_date: datetime, end_date: datetime
) -> list[GoProMediaSearchItem]:
    """Collect media rows across pages up to ``max_items``.

    Args:
        start_date: Capture range start.
        end_date: Capture range end.

    Returns:
        Up to ``self.max_items`` ``GoProMediaSearchItem`` instances.

    Raises:
        RuntimeError: If used outside ``with GoProClient()`` on any underlying
            ``search`` call.
        requests.HTTPError: When any underlying ``search`` HTTP status is not
            successful.
        pydantic.ValidationError: If any underlying ``search`` JSON body does not
            match the model.
    """
    all_media: list[GoProMediaSearchItem] = []
    for page_result in self.iter_nonempty_search_pages(start_date, end_date):
        all_media.extend(page_result.embedded.media)
        if len(all_media) >= self.max_items:
            break
    return all_media[: self.max_items]

get_download_url(media_items)

Resolve download assets for each search row.

Parameters:

Name Type Description Default
media_items list[GoProMediaSearchItem]

One or more media rows (typically from search).

 required

Returns:

Type Description
dict[str, DownloadAsset]

Merged mapping of output filename to file or variation metadata.

Raises:

Type Description
NoVariationsError

For video items with no variations.
RuntimeError

If used outside with GoProClient() on any underlying download call.
HTTPError

When any underlying download HTTP status is not successful.
ValidationError

If any underlying download JSON body does not match the model.
Source code in gopro_api/client.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def get_download_url(
    self, media_items: list[GoProMediaSearchItem]
) -> dict[str, DownloadAsset]:
    """Resolve download assets for each search row.

    Args:
        media_items: One or more media rows (typically from search).

    Returns:
        Merged mapping of output filename to file or variation metadata.

    Raises:
        NoVariationsError: For video items with no variations.
        RuntimeError: If used outside ``with GoProClient()`` on any underlying
            ``download`` call.
        requests.HTTPError: When any underlying ``download`` HTTP status is not
            successful.
        pydantic.ValidationError: If any underlying ``download`` JSON body does not
            match the model.
    """
    assets: dict[str, DownloadAsset] = {}
    for item in media_items:
        result = self._api.download(item.id)
        assets.update(
            pull_assets_for_response(
                result,
                target_height=self.prefer_height,
                target_width=self.prefer_width,
            )
        )
    return assets

download_url_to_path(url, dest_path)

Download a CDN URL to a local path.

Uses a one-shot requests.get because CDN hosts differ from api.gopro.com.

Parameters:

Name Type Description Default
url str

Fully qualified HTTPS URL from download metadata.

 required
dest_path str

Filesystem path for the response body.

 required

Raises:

Type Description
HTTPError

When the HTTP status is not successful.
OSError

If the destination cannot be written.
Source code in gopro_api/client.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def download_url_to_path(self, url: str, dest_path: str) -> None:
    """Download a CDN URL to a local path.

    Uses a one-shot ``requests.get`` because CDN hosts differ from
    ``api.gopro.com``.

    Args:
        url: Fully qualified HTTPS URL from download metadata.
        dest_path: Filesystem path for the response body.

    Raises:
        requests.HTTPError: When the HTTP status is not successful.
        OSError: If the destination cannot be written.
    """
    response = requests.get(url, timeout=self._timeout)
    response.raise_for_status()
    dest_dir = os.path.dirname(dest_path)
    if dest_dir:
        os.makedirs(dest_dir, exist_ok=True)
    with open(dest_path, "wb") as out_file:
        out_file.write(response.content)

AsyncGoProClient

gopro_api.client.AsyncGoProClient

High-level async client for GoPro cloud media.

Wraps AsyncGoProAPI via composition and mirrors GoProClient using async/await and aiohttp. Use as an async context manager; the underlying aiohttp.ClientSession is opened and closed for you.

Source code in gopro_api/client.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
class AsyncGoProClient:
    """High-level async client for GoPro cloud media.

    Wraps ``AsyncGoProAPI`` via composition and mirrors ``GoProClient`` using
    ``async/await`` and ``aiohttp``. Use as an async context manager; the
    underlying ``aiohttp.ClientSession`` is opened and closed for you.
    """

    def __init__(
        self,
        access_token: str | None = None,
        timeout: float = 10.0,
        *,
        page_size: int = 1000,
        max_items: int = 1,
        prefer_height: int | None = None,
        prefer_width: int | None = None,
    ) -> None:
        """Create an async high-level client.

        Args:
            access_token: ``gp_access_token`` cookie value; defaults to
                ``gopro_api.config.GP_ACCESS_TOKEN``.
            timeout: Total ``aiohttp`` client timeout in seconds.
            page_size: Default page size for ``iter_nonempty_search_pages``.
            max_items: Maximum rows returned by ``list_media_items``.
            prefer_height: Preferred video height in pixels for ``get_download_url``.
            prefer_width: Preferred video width in pixels for ``get_download_url``.
        """
        self._api = AsyncGoProAPI(access_token=access_token, timeout=timeout)
        self.page_size = page_size
        self.max_items = max_items
        self.prefer_height = prefer_height
        self.prefer_width = prefer_width

    async def __aenter__(self) -> "AsyncGoProClient":
        """Enter the underlying ``AsyncGoProAPI`` context.

        Returns:
            ``self``.
        """
        await self._api.__aenter__()
        return self

    async def __aexit__(self, *exc: object) -> None:
        """Exit the underlying ``AsyncGoProAPI`` context."""
        await self._api.__aexit__(*exc)

    # ------------------------------------------------------------------
    # Low-level proxies
    # ------------------------------------------------------------------

    async def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchResponse:
        """Run a single media search request.

        Args:
            params: Query parameters for ``GET /media/search``.

        Returns:
            Parsed search response.

        Raises:
            RuntimeError: If used outside ``async with AsyncGoProClient()``.
            aiohttp.ClientResponseError: When ``raise_for_status`` fails.
            pydantic.ValidationError: If the JSON body does not match the model.
        """
        return await self._api.search(params)

    async def download(self, media_id: str) -> GoProMediaDownloadResponse:
        """Fetch download metadata for one media id.

        Args:
            media_id: Cloud library identifier.

        Returns:
            Parsed download metadata response.

        Raises:
            RuntimeError: If used outside ``async with AsyncGoProClient()``.
            aiohttp.ClientResponseError: When ``raise_for_status`` fails.
            pydantic.ValidationError: If the JSON body does not match the model.
        """
        return await self._api.download(media_id)

    # ------------------------------------------------------------------
    # High-level helpers
    # ------------------------------------------------------------------

    async def iter_nonempty_search_pages(
        self,
        start_date: datetime,
        end_date: datetime,
        *,
        per_page: int | None = None,
        start_page: int = 1,
    ) -> AsyncIterator[GoProMediaSearchResponse]:
        """Yield search pages until one returns an empty ``_embedded.media``.

        Args:
            start_date: Capture range start (inclusive semantics per API).
            end_date: Capture range end.
            per_page: Items per page; defaults to ``self.page_size``.
            start_page: First page number to request (1-indexed).

        Yields:
            Each non-empty ``GoProMediaSearchResponse`` page.
        """
        page = start_page
        size = per_page if per_page is not None else self.page_size
        while True:
            params = GoProMediaSearchParams(
                captured_range=CapturedRange(start=start_date, end=end_date),
                page=page,
                per_page=size,
            )
            result = await self._api.search(params)
            if not result.embedded.media:
                return
            yield result
            page += 1

    async def list_media_items(
        self, start_date: datetime, end_date: datetime
    ) -> list[GoProMediaSearchItem]:
        """Collect media rows across pages up to ``max_items``.

        Args:
            start_date: Capture range start.
            end_date: Capture range end.

        Returns:
            Up to ``self.max_items`` ``GoProMediaSearchItem`` instances.

        Raises:
            RuntimeError: If used outside ``async with AsyncGoProClient()`` on any
                underlying ``search`` call.
            aiohttp.ClientResponseError: When any underlying ``search`` raises for
                status.
            pydantic.ValidationError: If any underlying ``search`` JSON body does not
                match the model.
        """
        all_media: list[GoProMediaSearchItem] = []
        async for page_result in self.iter_nonempty_search_pages(start_date, end_date):
            all_media.extend(page_result.embedded.media)
            if len(all_media) >= self.max_items:
                break
        return all_media[: self.max_items]

    async def get_download_url(
        self, media_items: list[GoProMediaSearchItem]
    ) -> dict[str, DownloadAsset]:
        """Resolve download assets for each search row in parallel.

        Args:
            media_items: One or more media rows (typically from search).

        Returns:
            Merged mapping of output filename to file or variation metadata.

        Raises:
            NoVariationsError: For video items with no variations.
            RuntimeError: If used outside ``async with AsyncGoProClient()`` on any
                underlying ``download`` call.
            aiohttp.ClientResponseError: When any underlying ``download`` raises for
                status.
            pydantic.ValidationError: If any underlying ``download`` JSON body does not
                match the model.
        """
        results: list[GoProMediaDownloadResponse] = await asyncio.gather(
            *(self._api.download(item.id) for item in media_items)
        )
        assets: dict[str, DownloadAsset] = {}
        for result in results:
            assets.update(
                pull_assets_for_response(
                    result,
                    target_height=self.prefer_height,
                    target_width=self.prefer_width,
                )
            )
        return assets

    async def download_url_to_path(self, url: str, dest_path: str) -> None:
        """Download a CDN URL to a local path.

        Opens a dedicated ``aiohttp.ClientSession`` without ``base_url`` so CDN
        hosts work. The body is read fully into memory, then written via
        ``asyncio.to_thread``.

        Args:
            url: Fully qualified HTTPS URL from download metadata.
            dest_path: Filesystem path for the response body.

        Raises:
            aiohttp.ClientResponseError: When ``raise_for_status`` fails.
            OSError: If the destination cannot be written.
        """
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                resp.raise_for_status()
                data = await resp.read()

        dest_dir = os.path.dirname(dest_path)
        if dest_dir:
            os.makedirs(dest_dir, exist_ok=True)
        await asyncio.to_thread(write_bytes, dest_path, data)

__init__(access_token=None, timeout=10.0, *, page_size=1000, max_items=1, prefer_height=None, prefer_width=None)

Create an async high-level client.

Parameters:

Name Type Description Default
access_token str | None

gp_access_token cookie value; defaults to gopro_api.config.GP_ACCESS_TOKEN.

 None
timeout float

Total aiohttp client timeout in seconds.

 10.0
page_size int

Default page size for iter_nonempty_search_pages.

 1000
max_items int

Maximum rows returned by list_media_items.

 1
prefer_height int | None

Preferred video height in pixels for get_download_url.

 None
prefer_width int | None

Preferred video width in pixels for get_download_url.

 None
Source code in gopro_api/client.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def __init__(
    self,
    access_token: str | None = None,
    timeout: float = 10.0,
    *,
    page_size: int = 1000,
    max_items: int = 1,
    prefer_height: int | None = None,
    prefer_width: int | None = None,
) -> None:
    """Create an async high-level client.

    Args:
        access_token: ``gp_access_token`` cookie value; defaults to
            ``gopro_api.config.GP_ACCESS_TOKEN``.
        timeout: Total ``aiohttp`` client timeout in seconds.
        page_size: Default page size for ``iter_nonempty_search_pages``.
        max_items: Maximum rows returned by ``list_media_items``.
        prefer_height: Preferred video height in pixels for ``get_download_url``.
        prefer_width: Preferred video width in pixels for ``get_download_url``.
    """
    self._api = AsyncGoProAPI(access_token=access_token, timeout=timeout)
    self.page_size = page_size
    self.max_items = max_items
    self.prefer_height = prefer_height
    self.prefer_width = prefer_width

__aenter__() async

Enter the underlying AsyncGoProAPI context.

Returns:

Type Description
'AsyncGoProClient'

self.
Source code in gopro_api/client.py 
272
273
274
275
276
277
278
279
async def __aenter__(self) -> "AsyncGoProClient":
    """Enter the underlying ``AsyncGoProAPI`` context.

    Returns:
        ``self``.
    """
    await self._api.__aenter__()
    return self

__aexit__(*exc) async

Exit the underlying AsyncGoProAPI context.

Source code in gopro_api/client.py 
281
282
283
async def __aexit__(self, *exc: object) -> None:
    """Exit the underlying ``AsyncGoProAPI`` context."""
    await self._api.__aexit__(*exc)

search(params) async

Run a single media search request.

Parameters:

Name Type Description Default
params GoProMediaSearchParams

Query parameters for GET /media/search.

 required

Returns:

Type Description
GoProMediaSearchResponse

Parsed search response.

Raises:

Type Description
RuntimeError

If used outside async with AsyncGoProClient().
ClientResponseError

When raise_for_status fails.
ValidationError

If the JSON body does not match the model.
Source code in gopro_api/client.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
async def search(self, params: GoProMediaSearchParams) -> GoProMediaSearchResponse:
    """Run a single media search request.

    Args:
        params: Query parameters for ``GET /media/search``.

    Returns:
        Parsed search response.

    Raises:
        RuntimeError: If used outside ``async with AsyncGoProClient()``.
        aiohttp.ClientResponseError: When ``raise_for_status`` fails.
        pydantic.ValidationError: If the JSON body does not match the model.
    """
    return await self._api.search(params)

download(media_id) async

Fetch download metadata for one media id.

Parameters:

Name Type Description Default
media_id str

Cloud library identifier.

 required

Returns:

Type Description
GoProMediaDownloadResponse

Parsed download metadata response.

Raises:

Type Description
RuntimeError

If used outside async with AsyncGoProClient().
ClientResponseError

When raise_for_status fails.
ValidationError

If the JSON body does not match the model.
Source code in gopro_api/client.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
async def download(self, media_id: str) -> GoProMediaDownloadResponse:
    """Fetch download metadata for one media id.

    Args:
        media_id: Cloud library identifier.

    Returns:
        Parsed download metadata response.

    Raises:
        RuntimeError: If used outside ``async with AsyncGoProClient()``.
        aiohttp.ClientResponseError: When ``raise_for_status`` fails.
        pydantic.ValidationError: If the JSON body does not match the model.
    """
    return await self._api.download(media_id)

iter_nonempty_search_pages(start_date, end_date, *, per_page=None, start_page=1) async

Yield search pages until one returns an empty _embedded.media.

Parameters:

Name Type Description Default
start_date datetime

Capture range start (inclusive semantics per API).

 required
end_date datetime

Capture range end.

 required
per_page int | None

Items per page; defaults to self.page_size.

 None
start_page int

First page number to request (1-indexed).

 1

Yields:

Type Description
AsyncIterator[GoProMediaSearchResponse]

Each non-empty GoProMediaSearchResponse page.
Source code in gopro_api/client.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
async def iter_nonempty_search_pages(
    self,
    start_date: datetime,
    end_date: datetime,
    *,
    per_page: int | None = None,
    start_page: int = 1,
) -> AsyncIterator[GoProMediaSearchResponse]:
    """Yield search pages until one returns an empty ``_embedded.media``.

    Args:
        start_date: Capture range start (inclusive semantics per API).
        end_date: Capture range end.
        per_page: Items per page; defaults to ``self.page_size``.
        start_page: First page number to request (1-indexed).

    Yields:
        Each non-empty ``GoProMediaSearchResponse`` page.
    """
    page = start_page
    size = per_page if per_page is not None else self.page_size
    while True:
        params = GoProMediaSearchParams(
            captured_range=CapturedRange(start=start_date, end=end_date),
            page=page,
            per_page=size,
        )
        result = await self._api.search(params)
        if not result.embedded.media:
            return
        yield result
        page += 1

list_media_items(start_date, end_date) async

Collect media rows across pages up to max_items.

Parameters:

Name Type Description Default
start_date datetime

Capture range start.

 required
end_date datetime

Capture range end.

 required

Returns:

Type Description
list[GoProMediaSearchItem]

Up to self.max_items GoProMediaSearchItem instances.

Raises:

Type Description
RuntimeError

If used outside async with AsyncGoProClient() on any underlying search call.
ClientResponseError

When any underlying search raises for status.
ValidationError

If any underlying search JSON body does not match the model.
Source code in gopro_api/client.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
async def list_media_items(
    self, start_date: datetime, end_date: datetime
) -> list[GoProMediaSearchItem]:
    """Collect media rows across pages up to ``max_items``.

    Args:
        start_date: Capture range start.
        end_date: Capture range end.

    Returns:
        Up to ``self.max_items`` ``GoProMediaSearchItem`` instances.

    Raises:
        RuntimeError: If used outside ``async with AsyncGoProClient()`` on any
            underlying ``search`` call.
        aiohttp.ClientResponseError: When any underlying ``search`` raises for
            status.
        pydantic.ValidationError: If any underlying ``search`` JSON body does not
            match the model.
    """
    all_media: list[GoProMediaSearchItem] = []
    async for page_result in self.iter_nonempty_search_pages(start_date, end_date):
        all_media.extend(page_result.embedded.media)
        if len(all_media) >= self.max_items:
            break
    return all_media[: self.max_items]

get_download_url(media_items) async

Resolve download assets for each search row in parallel.

Parameters:

Name Type Description Default
media_items list[GoProMediaSearchItem]

One or more media rows (typically from search).

 required

Returns:

Type Description
dict[str, DownloadAsset]

Merged mapping of output filename to file or variation metadata.

Raises:

Type Description
NoVariationsError

For video items with no variations.
RuntimeError

If used outside async with AsyncGoProClient() on any underlying download call.
ClientResponseError

When any underlying download raises for status.
ValidationError

If any underlying download JSON body does not match the model.
Source code in gopro_api/client.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
async def get_download_url(
    self, media_items: list[GoProMediaSearchItem]
) -> dict[str, DownloadAsset]:
    """Resolve download assets for each search row in parallel.

    Args:
        media_items: One or more media rows (typically from search).

    Returns:
        Merged mapping of output filename to file or variation metadata.

    Raises:
        NoVariationsError: For video items with no variations.
        RuntimeError: If used outside ``async with AsyncGoProClient()`` on any
            underlying ``download`` call.
        aiohttp.ClientResponseError: When any underlying ``download`` raises for
            status.
        pydantic.ValidationError: If any underlying ``download`` JSON body does not
            match the model.
    """
    results: list[GoProMediaDownloadResponse] = await asyncio.gather(
        *(self._api.download(item.id) for item in media_items)
    )
    assets: dict[str, DownloadAsset] = {}
    for result in results:
        assets.update(
            pull_assets_for_response(
                result,
                target_height=self.prefer_height,
                target_width=self.prefer_width,
            )
        )
    return assets

download_url_to_path(url, dest_path) async

Download a CDN URL to a local path.

Opens a dedicated aiohttp.ClientSession without base_url so CDN hosts work. The body is read fully into memory, then written via asyncio.to_thread.

Parameters:

Name Type Description Default
url str

Fully qualified HTTPS URL from download metadata.

 required
dest_path str

Filesystem path for the response body.

 required

Raises:

Type Description
ClientResponseError

When raise_for_status fails.
OSError

If the destination cannot be written.
Source code in gopro_api/client.py 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
async def download_url_to_path(self, url: str, dest_path: str) -> None:
    """Download a CDN URL to a local path.

    Opens a dedicated ``aiohttp.ClientSession`` without ``base_url`` so CDN
    hosts work. The body is read fully into memory, then written via
    ``asyncio.to_thread``.

    Args:
        url: Fully qualified HTTPS URL from download metadata.
        dest_path: Filesystem path for the response body.

    Raises:
        aiohttp.ClientResponseError: When ``raise_for_status`` fails.
        OSError: If the destination cannot be written.
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            resp.raise_for_status()
            data = await resp.read()

    dest_dir = os.path.dirname(dest_path)
    if dest_dir:
        os.makedirs(dest_dir, exist_ok=True)
    await asyncio.to_thread(write_bytes, dest_path, data)