# Google Drive connector

Source: /docs/connectors/google-drive.md

Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See [Custom API](/docs/connectors/custom-api) for the shared SDK pattern and document-upload workflow.

## What this connector is for [#what-this-connector-is-for]

Cloud file repository with file metadata, folder navigation, permissions, and content.

A connector owns access to the source. Separate syncs shape selected resources into tables for analysis and downstream transformations using the [Python SDK](/docs/reference/python/connectors-and-syncs).

## Data and tables you can pull [#data-and-tables-you-can-pull]

These are representative dataset families to consider, not a claim that the worked example implements every resource. Access, fields, and availability depend on the source account, product edition, and permissions. Use the [vendor reference](https://developers.google.com/workspace/drive/api/reference/rest/v3/files/list) to select concrete resources.

| Suggested table | One row represents |
| --- | --- |
| `files` | a file |
| `drives` | a drive |
| `permissions` | a permission |
| `revisions` | a revision |
| `changes` | a change |

### Relationships and extraction considerations [#relationships-and-extraction-considerations]

Separate file metadata from file content. Permissions determine visible files, including shared-drive access. Content extraction needs download/export handling in addition to the metadata listing shown here.

## Connecting to Google Drive [#connecting-to-google-drive]

Uses Google OAuth client credentials and a refresh token authorized for
`https://www.googleapis.com/auth/drive.metadata.readonly` (or a broader read scope).
Lists accessible file metadata, including shared-drive items. Does not download
file contents.

All credentials arrive through `config`: `client_id`, `client_secret`, and
`refresh_token` are required. `access_token` and `expires_at` (Unix seconds) are
optional cached values. Obtain the refresh token through Google's consent flow
with offline access before running this connector.

Inside `client()`, a request hook checks the cached token before every API
request, including subsequent pages. It reuses tokens until 60 seconds before
expiry, then exchanges the refresh token at Google's token endpoint. The same
HTTP client stays alive. A separate token request avoids recursively invoking the
hook, and a lock prevents simultaneous refreshes within that client.

Refreshed token values and expiry are written to `connector.config`. The platform
must persist a replacement refresh token if one is returned; changing this
dictionary alone does not save credentials across runs. Checkpoint state is never
used for authentication.

Additional resources may require additional scopes. Connection settings belong to the connector; dataset filters and resource selection belong to each sync. See the [CLI workflow](/docs/reference/cli/overview) for registration and configuration.

### Connection settings used by the example [#connection-settings-used-by-the-example]

| Setting | Requirement | Purpose |
| --- | --- | --- |
| `client_id` | Required | Google OAuth client ID used to obtain the refresh token. |
| `client_secret` | Required; secret | Secret for the Google OAuth client. |
| `refresh_token` | Required; secret | Google refresh token obtained with offline access and drive.metadata.readonly or a broader authorized scope. |
| `access_token` | Optional; secret | Optional cached token. If absent, the connector refreshes before the first API request. |
| `expires_at` | Optional | Optional Unix timestamp in seconds. A missing expiry triggers refresh; tokens are refreshed within 60 seconds of expiry. |

### requirements.in [#requirementsin]

Save these dependencies in `requirements.in` alongside `connector.py`; the example includes packages used by its sync. Install the published `vortexify` package for local development. Keep sync-specific dependencies in the sync's own `requirements.in` when splitting the implementation into separate resources.

```text
vortexify==0.1.1
httpx>=0.27,<1
pandas>=2,<3
```

No separately installed database driver is needed by this HTTP example.

## Worked example: sync `files` [#worked-example-sync-files]

**files** — Full read using the API's native pagination.

The complete files below implement only this example resource. They illustrate the current SDK signatures, a separate preview, bounded DataFrame output, and source-specific extraction. They do not establish complete source coverage. Live credentials and platform ingestion have not been used to validate this guide.

With default paths, place the connector files in `connectors/google-drive/` and sync files in `connectors/google-drive/syncs/files/`. Use your project configuration if it specifies different paths.

### Connection schema: `config.json` [#connection-schema-configjson]

This file contains a JSON Schema, not actual credential values. Provide those through platform configuration.

```json
{
  "type": "object",
  "properties": {
    "client_id": {
      "type": "string",
      "title": "OAuth Client ID",
      "description": "Google OAuth client ID used to obtain the refresh token.",
      "minLength": 1,
      "order": 0
    },
    "client_secret": {
      "type": "string",
      "title": "OAuth Client Secret",
      "description": "Secret for the Google OAuth client.",
      "minLength": 1,
      "x-vortexify-secret": true,
      "order": 1
    },
    "refresh_token": {
      "type": "string",
      "title": "Refresh Token",
      "description": "Google refresh token obtained with offline access and drive.metadata.readonly or a broader authorized scope.",
      "minLength": 1,
      "x-vortexify-secret": true,
      "order": 2
    },
    "access_token": {
      "type": "string",
      "title": "Cached Access Token",
      "description": "Optional cached token. If absent, the connector refreshes before the first API request.",
      "minLength": 1,
      "x-vortexify-secret": true,
      "order": 3
    },
    "expires_at": {
      "type": "number",
      "title": "Access Token Expiry",
      "description": "Optional Unix timestamp in seconds. A missing expiry triggers refresh; tokens are refreshed within 60 seconds of expiry.",
      "minimum": 0,
      "order": 4
    }
  },
  "required": [
    "client_id",
    "client_secret",
    "refresh_token"
  ]
}
```

### Connector: `connector.py` [#connector-connectorpy]

```python
from math import isfinite
from threading import Lock
from time import time
import httpx
from vortexify import Column, Connector, DiscoveredTable


class GoogleDriveConnector(Connector):
    vendor_domain = "google.com"

    def client(self):
        if self._client is None or self._client.is_closed:
            config = self.config
            refresh_lock = Lock()

            def authenticate(request: httpx.Request) -> None:
                with refresh_lock:
                    if (
                        not config.get("access_token")
                        or config.get("expires_at", 0) <= time() + 60
                    ):
                        requested_at = time()
                        response = httpx.post(
                            "https://oauth2.googleapis.com/token",
                            data={
                                "grant_type": "refresh_token",
                                "refresh_token": config["refresh_token"],
                                "client_id": config["client_id"],
                                "client_secret": config["client_secret"],
                            },
                            timeout=30,
                        )
                        response.raise_for_status()
                        tokens = response.json()
                        access_token = tokens["access_token"]
                        expires_in = float(tokens["expires_in"])
                        if not isinstance(access_token, str) or not access_token:
                            raise ValueError("Google returned an invalid access token")
                        if not isfinite(expires_in) or expires_in <= 0:
                            raise ValueError(
                                "Google returned an invalid token lifetime"
                            )
                        updated = {
                            "access_token": access_token,
                            "expires_at": requested_at + expires_in,
                        }
                        if tokens.get("refresh_token"):
                            updated["refresh_token"] = tokens["refresh_token"]
                        config.update(updated)
                    request.headers["Authorization"] = (
                        "Bearer " + config["access_token"]
                    )

            self._client = httpx.Client(
                base_url="https://www.googleapis.com/drive/v3/",
                headers={"Accept": "application/json"},
                event_hooks={"request": [authenticate]},
                timeout=60,
                follow_redirects=False,
            )
        return self._client

    def check(self) -> bool:
        response = self.client().get(
            "files", params={"pageSize": 1, "fields": "files(id)"}
        )
        response.raise_for_status()
        return True

    def discover(self) -> list[DiscoveredTable]:
        return [
            DiscoveredTable(
                name="files",
                schema=[
                    Column(name="id", type="string"),
                    Column(name="name", type="string"),
                    Column(name="mime_type", type="string"),
                    Column(name="modified_time", type="timestamp"),
                    Column(name="size", type="string"),
                    Column(name="document", type="string"),
                ],
            )
        ]

    def close(self) -> None:
        if self._client is not None:
            try:
                self._client.close()
            finally:
                self._client = None
```

### Sync schema: `syncs/files/config.json` [#sync-schema-syncsfilesconfigjson]

```json
{
  "type": "object",
  "properties": {},
  "required": []
}
```

### Sync: `syncs/files/sync.py` [#sync-syncsfilessyncpy]

```python
from collections.abc import Iterator
from math import isfinite
import json
import pandas as pd
from vortexify import Sync


class GoogleDriveFilesSync(Sync):

    def preview(self) -> pd.DataFrame:
        cfg = self.connector.config
        page_size = 100
        path = "files"
        params = {
            "pageSize": page_size,
            "q": "trashed = false",
            "fields": "nextPageToken,files(id,name,mimeType,modifiedTime,size,parents)",
            "supportsAllDrives": "true",
            "includeItemsFromAllDrives": "true",
        }
        response = self.client.get(path, params=params)
        response.raise_for_status()
        payload = response.json()
        records = payload.get("files", [])
        if not isinstance(records, list) or any(
            (not isinstance(record, dict) for record in records)
        ):
            raise ValueError(
                "Unexpected API record shape; refusing a partial successful sync"
            )
        records = records[:100]
        rows = []
        for record in records:
            row = {
                "id": record.get("id"),
                "name": record.get("name"),
                "mime_type": record.get("mimeType"),
                "modified_time": record.get("modifiedTime"),
                "size": record.get("size"),
                "document": json.dumps(record, ensure_ascii=False, default=str),
            }
            rows.append(
                {
                    key: (
                        json.dumps(value, ensure_ascii=False, default=str)
                        if isinstance(value, (dict, list))
                        else value
                    )
                    for key, value in row.items()
                }
            )
        return pd.DataFrame(
            rows,
            columns=["id", "name", "mime_type", "modified_time", "size", "document"],
            dtype=object,
        )

    def extract(
        self, *, avg_row_size_bytes: float | None = None
    ) -> Iterator[pd.DataFrame]:
        memory_mib = self.compute.memory if self.compute else None
        memory_mib = memory_mib if memory_mib is not None and memory_mib > 0 else 1024
        target_bytes = min(16 * 1024 * 1024, max(1, memory_mib * 1024 * 1024 // 20))
        batch_size = 100
        if avg_row_size_bytes is not None:
            if not isfinite(avg_row_size_bytes) or avg_row_size_bytes <= 0:
                raise ValueError("avg_row_size_bytes must be finite and positive")
            batch_size = max(1, int(target_bytes / max(1.0, avg_row_size_bytes) / 4))
        batch_size = min(batch_size, 1000)
        cfg = self.connector.config
        page_size = batch_size
        path = "files"
        params = {
            "pageSize": page_size,
            "q": "trashed = false",
            "fields": "nextPageToken,files(id,name,mimeType,modifiedTime,size,parents)",
            "supportsAllDrives": "true",
            "includeItemsFromAllDrives": "true",
        }
        seen_pages = set()
        while True:
            response = self.client.get(path, params=params)
            response.raise_for_status()
            payload = response.json()
            records = payload.get("files", [])
            if not isinstance(records, list) or any(
                (not isinstance(record, dict) for record in records)
            ):
                raise ValueError(
                    "Unexpected API record shape; refusing a partial successful sync"
                )
            rows = []
            for record in records:
                row = {
                    "id": record.get("id"),
                    "name": record.get("name"),
                    "mime_type": record.get("mimeType"),
                    "modified_time": record.get("modifiedTime"),
                    "size": record.get("size"),
                    "document": json.dumps(record, ensure_ascii=False, default=str),
                }
                rows.append(
                    {
                        key: (
                            json.dumps(value, ensure_ascii=False, default=str)
                            if isinstance(value, (dict, list))
                            else value
                        )
                        for key, value in row.items()
                    }
                )
            for start in range(0, len(rows), batch_size):
                yield pd.DataFrame(
                    rows[start : start + batch_size],
                    columns=[
                        "id",
                        "name",
                        "mime_type",
                        "modified_time",
                        "size",
                        "document",
                    ],
                    dtype=object,
                )
            next_token = payload.get("nextPageToken")
            if not next_token:
                break
            marker = json.dumps(next_token, sort_keys=True)
            if marker in seen_pages:
                raise ValueError("API repeated a pagination token")
            seen_pages.add(marker)
            params["pageToken"] = next_token
```

### Running and extending the example [#running-and-extending-the-example]

Register the connector before the sync and commit each resource separately:

```sh
vortexify create connector google-drive
vortexify create sync google-drive/files
```

Edit the files, inspect `vortexify status`, and use the [CLI guide](/docs/reference/cli/overview) to commit and configure them. Registration alone does not run extraction or deploy the connector.

Preview must run with a fresh connector and disposable state. The extraction implementation determines full/incremental behavior; merely setting a mode does not implement change capture. API/driver failures propagate unless explicitly handled in the code. Check source limits and single-record/response sizes as well as output batch size.

## Sources and further reading [#sources-and-further-reading]

- [Vendor API and data reference](https://developers.google.com/workspace/drive/api/reference/rest/v3/files/list) — Resource definitions, authentication, fields, and source-specific limits. Check the selected resource's reference when extending the example.
- [Vortexify connector and sync contracts](/docs/reference/python/connectors-and-syncs) — Client ownership, discovery, preview isolation, and DataFrame extraction.
- [Vortexify CLI workflow](/docs/reference/cli/overview) — Project paths, resource registration, source commits, and credential configuration.