# Google BigQuery connector

Source: /docs/connectors/bigquery.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]

Google Cloud analytics warehouse for SQL queries, datasets, and tables.

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://cloud.google.com/bigquery/docs) to select concrete resources.

| Suggested table | One row represents |
| --- | --- |
| `datasets` | a dataset |
| `tables` | a table |
| `table_rows` | a table row |
| `query_rows` | a query result row |

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

Datasets and tables describe storage; table/query rows contain the business data. Query execution can incur charges and needs job permissions in addition to table read access.

## Connecting to Google BigQuery [#connecting-to-google-bigquery]

Uses a service account or application default credentials with BigQuery metadata and table data read permissions. Discovery lists real dataset tables and column schemas. list_rows reads table rows without running a SQL query; views require query jobs and are outside this example.

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 |
| --- | --- | --- |
| `project` | Required | Project |
| `service_account` | Optional; secret | Optional service-account JSON object; otherwise use application default credentials. |
| `dataset` | Required | Dataset |

### 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
google-cloud-bigquery>=3.25,<4
pandas>=2,<3
```

Google client libraries use the configured credentials and scopes. Installing the library does not grant project/account access; select a supported Python runtime and compatible library version.

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

**table_rows** — Reads the configured table via the BigQuery tabledata API; schema comes from the actual table.

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/bigquery/` and sync files in `connectors/bigquery/syncs/table_rows/`. 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": {
    "project": {
      "type": "string",
      "minLength": 1,
      "title": "Project",
      "order": 0
    },
    "service_account": {
      "type": "object",
      "title": "Service Account",
      "description": "Optional service-account JSON object; otherwise use application default credentials.",
      "x-vortexify-secret": true,
      "order": 1
    },
    "dataset": {
      "type": "string",
      "minLength": 1,
      "title": "Dataset",
      "order": 2
    }
  },
  "required": [
    "project",
    "dataset"
  ]
}
```

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

```python
from google.cloud.bigquery import Client
from vortexify import Column, Connector, DiscoveredTable


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

    def client(self):
        if self._client is None:
            if self.config.get("service_account"):
                self._client = Client.from_service_account_info(
                    self.config["service_account"], project=self.config["project"]
                )
            else:
                self._client = Client(project=self.config["project"])
        return self._client

    def check(self) -> bool:
        self.client().get_dataset(self.config["project"] + "." + self.config["dataset"])
        return True

    def discover(self) -> list[DiscoveredTable]:
        client = self.client()
        dataset = self.config["project"] + "." + self.config["dataset"]
        resources = []
        for entry in client.list_tables(dataset):
            if entry.table_type != "TABLE":
                continue
            table = client.get_table(entry.reference)
            resources.append(
                DiscoveredTable(
                    name=table.full_table_id.replace(":", "."),
                    schema=[
                        Column(
                            name=column.name,
                            type=column.field_type,
                            nullable=column.mode != "REQUIRED",
                        )
                        for column in table.schema
                    ],
                )
            )
        return resources

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

### Sync schema: `syncs/table_rows/config.json` [#sync-schema-syncstable_rowsconfigjson]

```json
{
  "type": "object",
  "properties": {
    "table": {
      "type": "string",
      "title": "Table",
      "description": "Table ID within the configured dataset.",
      "minLength": 1,
      "order": 0
    }
  },
  "required": [
    "table"
  ]
}
```

### Sync: `syncs/table_rows/sync.py` [#sync-syncstable_rowssyncpy]

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


class BigqueryTableRowsSync(Sync):

    def preview(self) -> pd.DataFrame:
        cfg = self.connector.config
        table = self.client.get_table(
            cfg["project"] + "." + cfg["dataset"] + "." + self.config["table"]
        )
        columns = [column.name for column in table.schema]
        records = self.client.list_rows(table, page_size=100, max_results=100)
        rows = [
            {
                key: (
                    json.dumps(value, default=str)
                    if isinstance(value, (dict, list))
                    else value
                )
                for key, value in dict(record).items()
            }
            for record in records
        ]
        return pd.DataFrame(rows, columns=columns, 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, 10000)
        cfg = self.connector.config
        table = self.client.get_table(
            cfg["project"] + "." + cfg["dataset"] + "." + self.config["table"]
        )
        columns = [column.name for column in table.schema]
        records = self.client.list_rows(table, page_size=batch_size)
        rows = []
        for record in records:
            rows.append(
                {
                    key: (
                        json.dumps(value, default=str)
                        if isinstance(value, (dict, list))
                        else value
                    )
                    for key, value in dict(record).items()
                }
            )
            if len(rows) == batch_size:
                yield pd.DataFrame(rows, columns=columns, dtype=object)
                rows = []
        if rows:
            yield pd.DataFrame(rows, columns=columns, dtype=object)
```

### 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 bigquery
vortexify create sync bigquery/table_rows
```

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://cloud.google.com/bigquery/docs) — 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.