gcloud-aio-bigquery
gcloud-aio-bigquery is an asyncio client over the Google BigQuery REST v2 API. It provides small Job, Table, and Dataset classes for queries, metadata, streaming inserts, load and copy jobs, plus a helper that converts BigQuery's nested row response into Python dictionaries. It shares authentication and HTTP-session machinery with gcloud-aio-auth. This is an independent REST wrapper, not Google's official google-cloud-bigquery client, and it intentionally returns mostly raw API dictionaries.
Choose it when native asyncio transport matters more than an extensive object model. Read the source-level defaults before writing data, especially legacy SQL, unknown-field handling, partial insert errors, and load-job truncation.
Use it if
- Your asyncio service must call BigQuery without running Google's synchronous client in worker threads
- You prefer thin REST-shaped dictionaries over a large object model and can read Google's REST API documentation
- You need streaming inserts, dataset and table metadata, query jobs, or Cloud Storage load jobs in one async package
- You already use other gcloud-aio clients and want shared token and aiohttp session behavior
- You want Google's supported client, richer query configuration objects, dataframe integrations, or BigQuery Storage API acceleration: use google-cloud-bigquery and its companion packages
- You expect complete task-based documentation: the package README says documentation is still being worked on and points readers to a smoke test, while many details require reading signatures or Google's REST reference
- You want safe convenience defaults: Job.insert_via_query defaults use_legacy_sql to true, Table.insert ignores unknown row fields by default, and load jobs default to WRITE_TRUNCATE
- You need automatic job waiting: Job.result checks once and raises OSError while work is pending, so polling, deadline, cancellation, and backoff are application responsibilities
- You want strongly modeled responses and errors: most methods accept and return dictionaries, and partial streaming failures can arrive in an insertErrors field even when the HTTP request itself succeeds
Setup reality
Install gcloud-aio-bigquery on Python 3.8 through 3.x below 4; version 7.1.0 depends on gcloud-aio-auth. In Google Cloud, Application Default Credentials can often supply a project and token, while local or non-Google environments normally need a service-account JSON file passed as service_file or authentication configured through gcloud-aio-auth. The identity needs BigQuery permissions for every operation you call, and load jobs also need access to the referenced Cloud Storage objects. Pass project explicitly when possible. Otherwise the token tries to discover it and raises a generic exception if it cannot. The client owns an HTTP session unless you supply one, so use async with or await close() to avoid unclosed-session warnings; sharing an aiohttp ClientSession or Token across clients reduces connection and token churn. The API is close to Google's JSON wire format. Job.query expects a query_request dictionary, create and patch expect table resource dictionaries, and results often need query_response_to_dict. That converter handles common scalar and record types but its source raises KeyError for unsupported DATE, DATETIME, GEOGRAPHY, and TIME fields, so it is not a universal decoder. Be explicit about useLegacySql: false for normal GoogleSQL. Do not rely on Job.insert_via_query's default, which is true in 7.1.0. Also note the dangerous write defaults: Table.insert_via_load uses WRITE_TRUNCATE, and streaming Table.insert defaults ignore_unknown=True. Always inspect insertErrors because a mixed success response can still use a successful HTTP status. Jobs are asynchronous; result() does one status check and does not sleep. Build a deadline-bound polling loop and preserve location for regional jobs. For emulator tests, BIGQUERY_EMULATOR_HOST changes the API root, omits authorization headers, and project selection falls back through BIGQUERY_PROJECT_ID, GOOGLE_CLOUD_PROJECT, then the literal dev value. That convenience does not make every emulator match production BigQuery behavior.
Patterns
Run a synchronous BigQuery query requestrun-query
from gcloud.aio.bigquery import Job, query_response_to_dict
async def fetch_users():
async with Job(project='my-project') as jobs:
response = await jobs.query({
'query': 'SELECT id, name FROM `my-project.app.users` LIMIT 100',
'useLegacySql': False,
})
return query_response_to_dict(response)Job.query takes the REST request dictionary directly; always set useLegacySql to false for GoogleSQL.
Send named query parametersparameterize-query
request = {
'query': 'SELECT * FROM `my-project.app.orders` WHERE status = @status',
'useLegacySql': False,
'parameterMode': 'NAMED',
'queryParameters': [{
'name': 'status',
'parameterType': {'type': 'STRING'},
'parameterValue': {'value': 'paid'},
}],
}
response = await jobs.query(request)The client does not build parameter objects for you; use BigQuery REST v2 field names exactly.
Start an asynchronous GoogleSQL jobstart-query-job
from gcloud.aio.bigquery import Disposition, Job
job = Job(project='my-project', location='US')
response = await job.insert_via_query(
'SELECT CURRENT_TIMESTAMP() AS now',
use_legacy_sql=False,
write_disposition=Disposition.WRITE_EMPTY,
)
print(response['jobReference']['jobId'])Pass use_legacy_sql=False explicitly because the 7.1.0 helper defaults to legacy SQL.
Poll a job with a deadlinepoll-job
import asyncio
async def wait_for_job(job, attempts=30):
for _ in range(attempts):
status = await job.get_job()
if status.get('status', {}).get('state') == 'DONE':
if 'errorResult' in status['status']:
raise RuntimeError(status['status']['errors'])
return status
await asyncio.sleep(1)
raise TimeoutError(job.job_id)Job.result() checks only once and raises OSError while pending, so production code needs its own bounded polling policy.
Insert streaming rows and check partial errorsstream-rows
from gcloud.aio.bigquery import Table
async with Table('analytics', 'events', project='my-project') as table:
response = await table.insert(
[{'event_id': 'evt-1', 'kind': 'signup'}],
ignore_unknown=False,
insert_id_fn=lambda row: row['event_id'],
)
if response.get('insertErrors'):
raise RuntimeError(response['insertErrors'])A request can partly succeed while returning insertErrors; a stable insert ID also helps BigQuery deduplicate retries.
Start a Parquet load jobload-from-storage
from gcloud.aio.bigquery import Disposition, SourceFormat, Table
table = Table('analytics', 'events', project='my-project')
job = await table.insert_via_load(
['gs://my-bucket/events/*.parquet'],
source_format=SourceFormat.PARQUET,
write_disposition=Disposition.WRITE_APPEND,
)
await wait_for_job(job)Set write_disposition explicitly; the helper default is WRITE_TRUNCATE, which replaces table contents.
Create a table from a REST resourcecreate-table
from gcloud.aio.bigquery import Table
resource = {
'schema': {'fields': [
{'name': 'event_id', 'type': 'STRING', 'mode': 'REQUIRED'},
{'name': 'created_at', 'type': 'TIMESTAMP', 'mode': 'NULLABLE'},
]},
}
async with Table('analytics', 'events', project='my-project') as table:
created = await table.create(resource)create mutates the supplied dictionary by adding tableReference before sending it.
List tables with REST paginationlist-tables
from gcloud.aio.bigquery import Dataset
async with Dataset('analytics', project='my-project') as dataset:
page = await dataset.list_tables(params={'maxResults': 100})
for table in page.get('tables', []):
print(table['tableReference']['tableId'])
next_token = page.get('nextPageToken')Pagination is not automatic; pass pageToken on the next request when nextPageToken is present.
Reuse an aiohttp sessionshare-http-session
from aiohttp import ClientSession
from gcloud.aio.bigquery import Dataset, Table
async with ClientSession() as session:
datasets = Dataset(project='my-project', session=session)
events = Table('analytics', 'events', project='my-project', session=session)
dataset_page = await datasets.list_datasets()
table_info = await events.get()When you supply the session, keep its lifetime outside the clients and close it only after all requests finish.
Point tests at a BigQuery emulatoruse-emulator
import os
from gcloud.aio.bigquery import Dataset
os.environ['BIGQUERY_EMULATOR_HOST'] = '127.0.0.1:9050'
os.environ['BIGQUERY_PROJECT_ID'] = 'test-project'
async with Dataset(project='test-project') as datasets:
response = await datasets.list_datasets()Emulator mode uses HTTP and omits authorization headers; set environment variables before constructing the client.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| google-cloud-bigquery | PyPI | You want Google's official client, richer models, dataframe helpers, and the best-supported default |
| google-cloud-bigquery-storage | PyPI | Your main problem is high-throughput table reads through the BigQuery Storage API |
| aiogoogle | PyPI | You need a generic async client across many Google discovery APIs and accept lower-level request construction |