# BigQuery connector

> **CocoIndex v1.** This page documents CocoIndex **v1** — a ground-up redesign from v0. When writing code, ignore any v0 flow-builder DSL or deprecated decorators.
>
> Source: https://cocoindex.io/docs/connectors/bigquery/ · Docs index: https://cocoindex.io/docs/llms.txt · Agent skill: https://cocoindex.io/docs/skill.md
>
> v0→v1 quick map — if you reach for these v0 symbols, stop and use the v1 form: `@cocoindex.flow_def`/`FlowBuilder` → `coco.App` + a `@coco.fn` main function; `add_collector()`/`collect()`/`export()` → declare target states (`declare_row`, `declare_file`); `cocoindex.sources/functions/targets.*` → connector APIs (`localfs.walk_dir`, `coco.ops.*`, `postgres.declare_table_target`). Full mapping + API reference: https://cocoindex.io/docs/skill.md.

The `bigquery` connector provides target state APIs for writing rows to BigQuery tables. CocoIndex tracks the rows that should exist and applies table creation, upserts, updates, and deletes incrementally.

```python
from cocoindex.connectors import bigquery
```

**Note — Install**
Install the optional BigQuery dependency before using this connector:

```bash
pip install cocoindex[bigquery]
```

## Connection setup

Create a `ContextKey[bigquery.ConnectionConfig]` to identify the BigQuery connection, then provide it in your lifespan:

**Note**
The key name is load-bearing across runs. It is the stable identity CocoIndex uses to track managed rows. See [ContextKey as stable identity](/docs/programming_guide/context#contextkey-as-stable-identity) before renaming.

```python
import os
from collections.abc import Iterator

import cocoindex as coco
from cocoindex.connectors import bigquery

BIGQUERY = coco.ContextKey[bigquery.ConnectionConfig]("bigquery")

@coco.lifespan
def coco_lifespan(builder: coco.EnvironmentBuilder) -> Iterator[None]:
    builder.provide(
        BIGQUERY,
        bigquery.ConnectionConfig(
            project=os.environ.get("BIGQUERY_PROJECT"),
            credentials_path=os.environ.get("GOOGLE_APPLICATION_CREDENTIALS") or None,
            location=os.environ.get("BIGQUERY_LOCATION"),
        ),
    )
    yield
```

If `credentials_path` is omitted, the Google client uses Application Default Credentials.

### ConnectionConfig

```python
@dataclass(frozen=True)
class ConnectionConfig:
    project: str | None = None
    credentials_path: str | None = None
    location: str | None = None
```

**Parameters:**

- `project` - Optional Google Cloud project used by the BigQuery client.
- `credentials_path` - Optional path to a service account JSON key file. If omitted, Application Default Credentials are used.
- `location` - Optional BigQuery job location, for example `US` or `EU`.

## As target

The `bigquery` connector provides target state APIs for writing rows to tables.

### Tables

Declares a table as a target state. Returns a `TableTarget` for declaring rows.

```python
def declare_table_target(
    db: ContextKey[ConnectionConfig],
    table_name: str,
    table_schema: TableSchema[RowT],
    *,
    dataset: str,
    project: str | None = None,
    managed_by: Literal["system", "user"] = "system",
) -> TableTarget[RowT, coco.PendingS]
```

**Parameters:**

- `db` - A `ContextKey[ConnectionConfig]` identifying the connection to use.
- `table_name` - Name of the table.
- `table_schema` - Schema definition including columns and primary key.
- `dataset` - BigQuery dataset name.
- `project` - Optional project that owns the target table. If omitted, BigQuery uses the client project.
- `managed_by` - Whether CocoIndex manages the table lifecycle (`"system"`) or assumes it exists (`"user"`).

When `managed_by="system"`, CocoIndex creates the dataset and table if needed. Table changes use BigQuery DDL, and row changes use BigQuery `MERGE` for upserts.

**Note**
BigQuery primary key constraints are not enforced by BigQuery. CocoIndex still uses the primary key list to reconcile rows and generate `MERGE` and `DELETE` statements.

### Rows

Once a `TableTarget` is resolved, declare rows to be upserted:

```python
def TableTarget.declare_row(
    self,
    *,
    row: RowT,
) -> None
```

**Parameters:**

- `row` - A row object (dict, dataclass, NamedTuple, or Pydantic model). Must include all primary key columns.

## Table schema: from Python class

Define the table structure using a Python class:

```python
from dataclasses import dataclass

@dataclass
class ProductRow:
    id: str
    name: str
    price: float
    metadata: dict[str, object]

schema = await bigquery.TableSchema.from_class(
    ProductRow,
    primary_key=["id"],
)
```

Python types are automatically mapped to BigQuery column types:

| Python Type | BigQuery Type |
|-------------|---------------|
| `bool` | `BOOL` |
| `int` | `INT64` |
| `float` | `FLOAT64` |
| `decimal.Decimal` | `NUMERIC` |
| `str` | `STRING` |
| `bytes` | `BYTES` |
| `uuid.UUID` | `STRING` |
| `datetime.date` | `DATE` |
| `datetime.time` | `TIME` |
| `datetime.datetime` | `TIMESTAMP` |
| `datetime.timedelta` | `FLOAT64` |
| `list`, `dict`, nested structs | `JSON` |

`JSON` values are JSON-serialized and written with `PARSE_JSON`.

### BigQueryType

Use `BigQueryType` to specify a custom BigQuery type and optional encoder:

```python
from dataclasses import dataclass
from typing import Annotated

from cocoindex.connectors.bigquery import BigQueryType

@dataclass
class ProductRow:
    id: Annotated[int, BigQueryType("NUMERIC")]
    embedding: Annotated[list[float], BigQueryType("ARRAY<FLOAT64>")]
```

You can also pass `column_overrides` when constructing the schema:

```python
schema = await bigquery.TableSchema.from_class(
    ProductRow,
    primary_key=["id"],
    column_overrides={
        "id": bigquery.BigQueryType("NUMERIC"),
    },
)
```

## Example

```python
from dataclasses import dataclass

@dataclass
class ProductRow:
    id: str
    name: str
    price: float
    metadata: dict[str, object]

async def declare_products(rows: list[ProductRow]) -> None:
    table = await bigquery.mount_table_target(
        BIGQUERY,
        table_name="product_index",
        table_schema=await bigquery.TableSchema.from_class(
            ProductRow,
            primary_key=["id"],
        ),
        project="my-project",
        dataset="analytics",
    )

    for row in rows:
        table.declare_row(row=row)
```

See `examples/bigquery_target` for a runnable project.

## Identifier handling

Dataset, table, and column names must be simple BigQuery identifiers containing letters, numbers, and underscores, and must not start with a number. The connector quotes identifiers when generating SQL.