Skip to content

GoPro Photos HTTP

Low-level aiohttp wrapper for the Google Photos Library v1 API, OAuth helpers, and Pydantic models.

API client

q2google.gphotos.api

Async client for Google Photos Library API v1 (uploads, batch create, media item lookup).

Use :class:GooglePhotosAPI only as an async context manager so an aiohttp session is opened and closed correctly.

GooglePhotosAPI(credentials: GooglePhotosOAuth, timeout: float = 600.0)

Thin aiohttp wrapper around selected Google Photos Library v1 endpoints.

Instantiate and use async with GooglePhotosAPI(...) as api to obtain a live session. HTTP error responses are surfaced via response.raise_for_status() (aiohttp client errors).

Create a client; the HTTP session starts on async context enter.

Parameters:

Name Type Description Default
credentials GooglePhotosOAuth

OAuth credentials carrying a valid access token.

 required
timeout float

Per-request total timeout in seconds. Each resumable chunk is one request; large uploads often need a generous value.

 600.0
Source code in q2google/gphotos/api.py 
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, credentials: GooglePhotosOAuth, timeout: float = 600.0) -> None:
    """Create a client; the HTTP session starts on async context enter.

    Args:
        credentials: OAuth credentials carrying a valid access token.
        timeout: Per-request total timeout in seconds. Each resumable chunk is one request;
            large uploads often need a generous value.
    """
    self._credentials = credentials
    self._token: str | None = None
    self._timeout = aiohttp.ClientTimeout(total=timeout)
    self._session: aiohttp.ClientSession | None = None

base_url: str property

Root URL for Library API v1 resources for this client.

__aenter__() -> GooglePhotosAPI async

Open the underlying aiohttp client session.

Source code in q2google/gphotos/api.py 
48
49
50
51
52
async def __aenter__(self) -> GooglePhotosAPI:
    """Open the underlying ``aiohttp`` client session."""
    logging.info("Creating aiohttp client session")
    self._session = aiohttp.ClientSession(timeout=self._timeout)
    return self

__aexit__(exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None) -> None async

Close the client session.

Parameters:

Name Type Description Default
exc_type type[BaseException] | None

Exception type if the context exited due to an error, else None.

 required
exc BaseException | None

Active exception instance when exiting with an error, else None.

 required
tb TracebackType | None

Traceback associated with exc, if any.

 required
Source code in q2google/gphotos/api.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Close the client session.

    Args:
        exc_type: Exception type if the context exited due to an error, else ``None``.
        exc: Active exception instance when exiting with an error, else ``None``.
        tb: Traceback associated with ``exc``, if any.
    """
    if self._session is not None:
        await self._session.close()
        self._session = None

create_media_item(media_item: MediaItemBatchCreateRequest) -> MediaItemBatchCreateResponse async

Call mediaItems:batchCreate with the given payload.

Parameters:

Name Type Description Default
media_item MediaItemBatchCreateRequest

Batch create request body.

 required

Returns:

Type Description
MediaItemBatchCreateResponse

API batch create response.
Source code in q2google/gphotos/api.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
async def create_media_item(self, media_item: MediaItemBatchCreateRequest) -> MediaItemBatchCreateResponse:
    """Call ``mediaItems:batchCreate`` with the given payload.

    Args:
        media_item: Batch create request body.

    Returns:
        API batch create response.
    """
    logging.info("Creating media item")
    session = self._get_session_or_raise()
    async with session.post(
        self.base_url + "mediaItems:batchCreate",
        headers={
            **self._auth_headers(),
            "Content-type": "application/json",
        },
        json=media_item.model_dump(),
    ) as response:
        logging.info(response.headers)
        response.raise_for_status()
        data = await response.json()
        logging.info(data)
        return MediaItemBatchCreateResponse.model_validate(data)

get_media_item(media_item_id: str) -> dict[str, Any] async

GET a single media item by id (mediaItems/{mediaItemId}).

Parameters:

Name Type Description Default
media_item_id str

Library media item id.

 required

Returns:

Type Description
dict[str, Any]

Parsed JSON object as returned by the API (shape matches the MediaItem resource).
Source code in q2google/gphotos/api.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
async def get_media_item(self, media_item_id: str) -> dict[str, Any]:
    """GET a single media item by id (``mediaItems/{mediaItemId}``).

    Args:
        media_item_id: Library media item id.

    Returns:
        Parsed JSON object as returned by the API (shape matches the MediaItem resource).
    """
    logging.info(f"Getting media item with ID: {media_item_id}")
    session = self._get_session_or_raise()
    url = self.base_url + f"mediaItems/{media_item_id}"
    async with session.get(
        url,
        headers={
            **self._auth_headers(),
            "Content-type": "application/json",
        },
    ) as response:
        response.raise_for_status()
        data = await response.json()
        logging.info(data)
        return data

init_upload_session(content_type: str, content_length: int) -> ResumableUploadSession async

Start a resumable upload; maps response headers to a session model.

Parameters:

Name Type Description Default
content_type str

MIME type of the file to upload.

 required
content_length int

Total byte size of the file.

 required

Returns:

Type Description
ResumableUploadSession

Parsed upload session metadata from response headers.
Source code in q2google/gphotos/api.py 
 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
async def init_upload_session(self, content_type: str, content_length: int) -> ResumableUploadSession:
    """Start a resumable upload; maps response headers to a session model.

    Args:
        content_type: MIME type of the file to upload.
        content_length: Total byte size of the file.

    Returns:
        Parsed upload session metadata from response headers.
    """
    logging.info("Initializing upload session")
    session = self._get_session_or_raise()
    headers = {
        **self._auth_headers(),
        "Content-Length": "0",
        "Content-Type": "application/octet-stream",
        "X-Goog-Upload-Command": "start",
        "X-Goog-Upload-Content-Type": content_type,
        "X-Goog-Upload-Protocol": "resumable",
        "X-Goog-Upload-Raw-Size": str(content_length),
    }
    async with session.post(self.base_url + "uploads", headers=headers) as response:
        logging.info(response.headers)
        logging.info(await response.text())
        response.raise_for_status()
        return ResumableUploadSession.model_validate(dict(response.headers))

query_upload_status(url: str) -> ResumableUploadSession async

Send an upload query command and return parsed header state.

Parameters:

Name Type Description Default
url str

Resumable upload URL to query.

 required

Returns:

Type Description
ResumableUploadSession

Parsed session metadata from response headers.
Source code in q2google/gphotos/api.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def query_upload_status(self, url: str) -> ResumableUploadSession:
    """Send an upload ``query`` command and return parsed header state.

    Args:
        url: Resumable upload URL to query.

    Returns:
        Parsed session metadata from response headers.
    """
    logging.info("Querying upload status")
    session = self._get_session_or_raise()
    async with session.post(
        url,
        headers={
            **self._auth_headers(),
            "Content-Length": str(0),
            "X-Goog-Upload-Command": "query",
        },
    ) as response:
        logging.info(response.headers)
        body = await response.text()
        logging.info(body)
        response.raise_for_status()
        return ResumableUploadSession.model_validate(dict(response.headers))

upload_chunk(url: str, command: str, offset: int, content: bytes) -> ResumableUploadSession async

POST one chunk of a resumable upload.

Parameters:

Name Type Description Default
url str

Resumable upload URL from a prior session.

 required
command str

Google resumable upload command (e.g. upload, finalize).

 required
offset int

Byte offset for this chunk.

 required
content bytes

Raw chunk bytes.

 required

Returns:

Type Description
ResumableUploadSession

Parsed session state; when the upload completes, upload_token may be set from
ResumableUploadSession

the response body.
Source code in q2google/gphotos/api.py 
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
async def upload_chunk(self, url: str, command: str, offset: int, content: bytes) -> ResumableUploadSession:
    """POST one chunk of a resumable upload.

    Args:
        url: Resumable upload URL from a prior session.
        command: Google resumable upload command (e.g. ``upload``, ``finalize``).
        offset: Byte offset for this chunk.
        content: Raw chunk bytes.

    Returns:
        Parsed session state; when the upload completes, ``upload_token`` may be set from
        the response body.
    """
    logging.info(f"Uploading chunk {offset} to {url}")
    session = self._get_session_or_raise()
    async with session.post(
        url,
        headers={
            **self._auth_headers(),
            "Content-Length": str(len(content)),
            "Content-Type": "application/octet-stream",
            "X-Goog-Upload-Command": command,
            "X-Goog-Upload-Offset": str(offset),
        },
        data=content,
    ) as response:
        logging.info(response.headers)
        body = await response.text()
        logging.info(body)
        response.raise_for_status()
        return ResumableUploadSession.model_validate(dict(response.headers, upload_token=body))

OAuth

q2google.gphotos.auth

OAuth helpers for Google Photos Library API using installed-app credentials.

GooglePhotosOAuth(client_secrets_file: str, scopes: list[PhotosScopes], token_file: str | None = None) dataclass

Load, refresh, or obtain Google OAuth credentials for Photos scopes.

Attributes:

Name Type Description
client_secrets_file str

Path to the Google OAuth client secrets JSON (installed app).
scopes list[PhotosScopes]

API scopes to request; values are sent to the authorization server.
token_file str | None

Optional path to persist the authorized user token; parent dirs are created on save. If omitted, tokens are only kept in memory.

authorize_interactive() -> Credentials

Run the local-server OAuth flow and optionally persist the token.

Returns:

Type Description
Credentials

Newly authorized credentials.
Source code in q2google/gphotos/auth.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def authorize_interactive(self) -> Credentials:
    """Run the local-server OAuth flow and optionally persist the token.

    Returns:
        Newly authorized credentials.
    """
    flow = InstalledAppFlow.from_client_secrets_file(
        self.client_secrets_file,
        scopes=[s.value for s in self.scopes],
    )
    creds = flow.run_local_server(open_browser=False)
    if self.token_file is not None:
        self._save_token(creds)
    return creds

ensure_credentials() -> Credentials

Return stored valid credentials, or complete an interactive authorization.

Returns:

Type Description
Credentials

Credentials that are valid for API calls (refreshed from disk when possible).

Raises:

Type Description
Various ``google_auth`` exceptions

If the OAuth flow or refresh fails in a way not handled by :meth:load_saved_credentials.
Source code in q2google/gphotos/auth.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def ensure_credentials(self) -> Credentials:
    """Return stored valid credentials, or complete an interactive authorization.

    Returns:
        Credentials that are valid for API calls (refreshed from disk when possible).

    Raises:
        Various ``google_auth`` exceptions: If the OAuth flow or refresh fails in a way not
            handled by :meth:`load_saved_credentials`.
    """
    saved = self.load_saved_credentials()
    if saved is not None and saved.valid:
        return saved
    return self.authorize_interactive()

load_saved_credentials() -> Credentials | None

Load credentials from disk and refresh if expired.

Returns:

Type Description
Credentials | None

Valid or refreshed credentials, or None if there is no token file,
Credentials | None

if RefreshError occurs during refresh (the token file is then removed),
Credentials | None

or if stored credentials cannot be used.
Source code in q2google/gphotos/auth.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def load_saved_credentials(self) -> Credentials | None:
    """Load credentials from disk and refresh if expired.

    Returns:
        Valid or refreshed credentials, or ``None`` if there is no token file,
        if ``RefreshError`` occurs during refresh (the token file is then removed),
        or if stored credentials cannot be used.
    """
    if self.token_file is None or not os.path.isfile(self.token_file):
        return None

    creds = Credentials.from_authorized_user_file(self.token_file)
    if creds.expired and creds.refresh_token:
        try:
            creds.refresh(Request())
            self._save_token(creds)
        except RefreshError:
            try:
                os.unlink(self.token_file)
            except OSError:
                pass
            return None
    return creds

Models

q2google.gphotos.models

Pydantic models and enums for Google Photos Library API payloads and upload headers.

Types mirror JSON resources and resumable-upload header fields.

MediaItem

Bases: BaseModel

A media item resource as returned by the Library API.

Examples:

>>> MediaItem(
...     id="ABC",
...     productUrl="https://photos.google.com/lr/photo/ABC",
...     mimeType="image/jpeg",
... )

baseUrl: str | None = None class-attribute instance-attribute

Base URL for requesting image bytes with size parameters, when present.

description: str | None = None class-attribute instance-attribute

User-visible description when present.

filename: str | None = None class-attribute instance-attribute

Filename when available.

id: str instance-attribute

Opaque media item identifier.

mediaMetadata: MediaMetadata | None = None class-attribute instance-attribute

Dimensions, timestamps, and other metadata.

mimeType: str instance-attribute

MIME type of the primary asset.

productUrl: str instance-attribute

URL to open the item in Google Photos.

MediaItemBatchCreateRequest

Bases: BaseModel

Request body for mediaItems:batchCreate.

Examples:

>>> MediaItemBatchCreateRequest(
...     newMediaItems=[NewMediaItem(simpleMediaItem=SimpleMediaItem(uploadToken="t"))],
... )

albumId: str | None = None class-attribute instance-attribute

If set, add created items to this album id.

newMediaItems: list[NewMediaItem] instance-attribute

Items to create from prior upload tokens.

MediaItemBatchCreateResponse

Bases: BaseModel

Response body from mediaItems:batchCreate.

Examples:

>>> MediaItemBatchCreateResponse(newMediaItemResults=[])

newMediaItemResults: list[NewMediaItemResult] instance-attribute

Parallel list of results aligned with the request newMediaItems order.

MediaMetadata

Bases: BaseModel

Subset of mediaMetadata returned on a media item; extra keys are preserved.

Examples:

>>> MediaMetadata(creationTime="2024-01-01T00:00:00Z", width=1920, height=1080)

creationTime: str | None = None class-attribute instance-attribute

ISO 8601 creation time when provided.

height: int | None = None class-attribute instance-attribute

Pixel height when provided.

width: int | None = None class-attribute instance-attribute

Pixel width when provided.

NewMediaItem

Bases: BaseModel

One new media item entry inside a batch create request.

Examples:

>>> NewMediaItem(simpleMediaItem=SimpleMediaItem(uploadToken="tok"))

description: str | None = None class-attribute instance-attribute

Optional description stored with the media item.

simpleMediaItem: SimpleMediaItem instance-attribute

Upload token and optional filename for this item.

NewMediaItemResult

Bases: BaseModel

Outcome for one entry in a mediaItems:batchCreate response.

Examples:

>>> NewMediaItemResult(
...     uploadToken="t",
...     status=Status(message="Success"),
... )

mediaItem: MediaItem | None = None class-attribute instance-attribute

Created media item when the call succeeded for this token.

status: Status instance-attribute

Success or error details for this token.

uploadToken: str instance-attribute

Upload token this result corresponds to.

PhotosScopes

Bases: str, Enum

OAuth 2.0 scope URLs accepted by the Google Photos Library API.

Examples:

>>> PhotosScopes.LIBRARY_APPENDONLY

LIBRARY_APPENDONLY = 'https://www.googleapis.com/auth/photoslibrary.appendonly' class-attribute instance-attribute

Create albums and upload media; cannot read or delete unrelated library content.

LIBRARY_EDIT_APP_CREATED = 'https://www.googleapis.com/auth/photoslibrary.edit.appcreateddata' class-attribute instance-attribute

Edit and delete media and albums created by this app.

LIBRARY_READONLY_APP_CREATED = 'https://www.googleapis.com/auth/photoslibrary.readonly.appcreateddata' class-attribute instance-attribute

Read media and albums created by this app only.

ResumableUploadSession

Bases: BaseModel

State parsed from Google resumable upload response headers (and optional body token).

Examples:

>>> ResumableUploadSession.model_validate(
...     {
...         "X-Goog-Upload-Status": "active",
...         "X-GUploader-UploadID": "upload-id",
...         "Date": "Mon, 01 Jan 2024 00:00:00 GMT",
...     }
... )

date: datetime = Field(alias='Date') class-attribute instance-attribute

Response Date header value, parsed to UTC-aware naive or as returned.

granularity: int | None = Field(alias='X-Goog-Upload-Chunk-Granularity', default=None) class-attribute instance-attribute

Preferred chunk size in bytes, if the server advertises one.

status: str = Field(alias='X-Goog-Upload-Status') class-attribute instance-attribute

Upload lifecycle status string from X-Goog-Upload-Status.

upload_id: str = Field(alias='X-GUploader-UploadID') class-attribute instance-attribute

Server upload identifier from X-GUploader-UploadID.

upload_token: str | None = None class-attribute instance-attribute

Raw upload token from the response body when the upload is complete.

upload_url: str | None = Field(alias='X-Goog-Upload-URL', default=None) class-attribute instance-attribute

URL to POST the next chunk or query, when provided by the server.

parse_date(v: str) -> datetime

Parse RFC 2822-style date strings from upload responses.

Parameters:

Name Type Description Default
v str

Date header value as returned by Google.

 required

Returns:

Type Description
datetime

Parsed datetime.
Source code in q2google/gphotos/models.py 
65
66
67
68
69
70
71
72
73
74
75
@field_validator("date", mode="before")
def parse_date(cls, v: str) -> datetime:
    """Parse RFC 2822-style date strings from upload responses.

    Args:
        v: Date header value as returned by Google.

    Returns:
        Parsed datetime.
    """
    return datetime.strptime(v, "%a, %d %b %Y %H:%M:%S %Z")

SimpleMediaItem

Bases: BaseModel

Minimal media reference for batch create (upload token plus optional filename).

Examples:

>>> SimpleMediaItem(uploadToken="token", fileName="photo.jpg")

fileName: str | None = None class-attribute instance-attribute

Suggested filename for the created item.

uploadToken: str instance-attribute

Upload token obtained after a successful resumable upload.

Status

Bases: BaseModel

Per-item status in a batch create response.

Examples:

>>> Status(message="Success")

code: int | None = None class-attribute instance-attribute

Optional numeric status code from the API.

details: list[str] | None = None class-attribute instance-attribute

Optional list of additional detail strings.

message: str instance-attribute

Human-readable status or error message.