Lazy BigQuery (zbq)

A lightweight, enhanced wrapper around Google Cloud BigQuery and Storage with Polars integration and seamless pandas compatibility. Simplifies querying and data operations with a unified interface, supporting read, write, insert, delete operations on BigQuery tables, and advanced file upload/download with pattern matching, parallel processing, and comprehensive error handling.

Table of Contents

• Features
• Installation
• Quick Start
  • BigQuery Operations
  • Storage Operations
• Parameter Substitution & Security
• Common Use Cases & Examples
• Async & Concurrency
• Scheduled Queries
• Advanced Configuration
• Error Handling
• Performance Tips
• API Reference
• Authentication & Setup

Key Features

BigQuery Operations

• Secure parameterization - Native @param_name syntax with SQL injection protection
• Type-safe parameters - Automatic type detection for strings, numbers, booleans, arrays
• Dual DataFrame support - Native Polars with seamless pandas conversion via .to_pandas()
• Dry run - dry_run() previews the final query with parameters substituted into literals, validated against BigQuery, with an estimate of bytes processed - no execution, no billing
• Free table preview - preview() samples rows via the tabledata.list API, so it scans no bytes and isn't billed
• Performance optimized - Built on Polars for speed, converts to pandas when needed
• Timeouts and typed errors - Per-call timeouts with a shared deadline budget, and Zbq* exceptions for authentication, configuration and operation failures
• Cost limit - max_bytes_billed on read() and execute() makes BigQuery refuse, without charge, a query that would bill more than the limit
• Explicit parameter types - TypedParam for typed NULLs, empty arrays, DATETIME and NUMERIC parameters
• Scheduled queries - Create, inspect, run and delete BigQuery scheduled queries with zscheduler (optional extra)

Storage Operations

• Single file uploads - Direct file upload with upload_file() or enhanced upload() auto-detection
• Read files to DataFrame - Read GCS files directly into Polars DataFrames with read() and read_iter()
• Multiple format support - CSV, Parquet, JSON, NDJSON with auto-detection
• Advanced pattern matching - Multiple include/exclude patterns, regex support, case-insensitive
• Parallel processing - Configurable thread pools for better performance
• Progress tracking - Built-in progress bars with tqdm and custom callbacks
• Dry-run mode - Preview operations without executing
• Retry logic - Automatic retry with exponential backoff for transient storage failures (timeouts, connection errors, 429 and 5xx). Permanent errors such as 403 and 404 fail immediately. BigQuery calls are not retried
• Detailed results - Comprehensive statistics including file counts, bytes transferred, duration, errors

Installation

uv add zbq

Upgrading to 2.2

2.2 adds features and fixes several bugs. Most upgrades need no changes, but these can affect existing code:

• ReadResult is removed from zbq and zbq.storage. Nothing returned it.
• timeout must be positive. timeout=0 used to fall back to the default silently; zero or a negative value now raises ValueError.
• Timeouts are detected by exception type, not by searching the error text for "timeout". An unrelated error that mentions the word is no longer turned into a TimeoutError.
• Storage retries only transient errors (timeouts, connection errors, 429 and 5xx). A 403, 404 or parse error is raised immediately instead of being retried. max_retries=0 now runs the operation once instead of doing nothing.
• upload(dry_run=True) on a single file now previews it instead of uploading it.
• @table and @table_name placeholders are replaced as whole words, so @table no longer alters @table_name or @tablex. The warning logged for table parameters is removed.
• The zbq console script is removed. It pointed at a function that never existed, so it could not run.
• Cancelling an async call cancels its BigQuery jobs, so they stop billing.

New in 2.2: max_bytes_billed (see Cost Limit), TypedParam (see Explicit Types with TypedParam), read_iter(on_error="raise"), and Scheduled Queries, which need uv add "zbq[scheduling]".

Upgrading to 2.1

2.1 tightens the API. Every change below fails loudly rather than silently changing behavior:

• query() is renamed to execute(). The arguments are the same apart from the dry-run flags.
• read() is always read-only. The read_only argument, the BigQueryHandler(read_only=...) / AsyncBigQueryHandler(read_only=...) constructor option, and the read_only attribute are gone. Run anything that modifies data with execute().
• dry_run= and dry_run_stats= are removed from read(), execute(), insert(), update(), and delete(). Call dry_run(query, parameters) instead; it always returns a DryRunResult.
• execute(), insert(), update(), and delete() return a JobResult (job_id, statement_type, num_dml_affected_rows, total_bytes_processed) instead of a one-row {"status", "job_id"} DataFrame. execute(..., return_dataframe=True) still returns a DataFrame.
• write() no longer prompts with input(). The warning argument is replaced by confirm: a truncate write raises ZbqOperationError unless confirm=True is passed.
• zbq no longer adds or removes loguru sinks. Its messages go wherever your application's loguru configuration sends them. The log_level argument on BigQueryHandler, AsyncBigQueryHandler, and StorageHandler, and the close() method, are removed. See Custom Logging.
• write() raises ValueError for a write_type other than "append" or "truncate", instead of treating anything unrecognized as an append.

2.0	2.1
zclient.query(sql)	zclient.execute(sql)
zclient.read(sql, read_only=False)	zclient.execute(sql, return_dataframe=True)
zclient.read(sql, dry_run=True, dry_run_stats=True)	zclient.dry_run(sql)
result["job_id"][0]	result.job_id
write(df, path, write_type="truncate", warning=False)	write(df, path, write_type="truncate", confirm=True)

Two more changes can't fail loudly, so check them before upgrading:

• Python 3.11 or newer is required. On Python 3.10, installers keep you on the previous release.
• The GOOGLE_CLOUD_PROJECT environment variable now takes priority over gcloud config. Before, it was almost never read. If you have it set to a different project than gcloud config get-value project, zbq now uses the environment variable's project. zbq also now follows gcloud config configurations activate, where it used to always read the default configuration. Pass project_id= explicitly if you need a fixed project. See Project Detection Order.

Get Started in 30 Seconds

from zbq import zclient, zstorage

# 1. Query BigQuery (returns Polars DataFrame)
df = zclient.read("SELECT * FROM `project.dataset.table` LIMIT 10")
print(df)

# 2. Upload a single file to Google Cloud Storage
result = zstorage.upload_file("./data.csv", "my-bucket/folder")
print(f"Uploaded: {result.uploaded_files} files")

# 3. Secure parameterized query
customers = zclient.read(
    "SELECT * FROM customers WHERE region = @region AND active = @active",
    parameters={"region": "US", "active": True}
)

# 4. Write data back to BigQuery (truncate requires explicit confirmation)
zclient.write(customers, "project.dataset.filtered_customers", write_type="truncate", confirm=True)

Quick Start

BigQuery Operations

from zbq import zclient

# Check/set project (optional - auto-detected by default)
print(f"Using project: {zclient.project_id}")
# zclient.project_id = "my-project-id"  # Override if needed

# Simple query (returns Polars DataFrame)
df = zclient.read("SELECT * FROM `project.dataset.table` LIMIT 1000")
print(df)

# Convert to pandas DataFrame if needed
pandas_df = zclient.read("SELECT * FROM `project.dataset.table` LIMIT 1000").to_pandas()
print(type(pandas_df))  # <class 'pandas.core.frame.DataFrame'>

# Peek at a table for free - uses tabledata.list, not a query job, so no bytes billed
sample_df = zclient.preview("project.dataset.table", max_results=10)
print(sample_df)

# Parameterized query (recommended) - works with both Polars and pandas
df = zclient.read(
    "SELECT * FROM @table_name WHERE region = @region LIMIT @limit",
    parameters={
        "table_name": "project.dataset.table",
        "region": "US", 
        "limit": 1000
    }
)  # Returns Polars DataFrame

# Same query as pandas DataFrame
pandas_df = zclient.read(
    "SELECT * FROM @table_name WHERE region = @region LIMIT @limit",
    parameters={
        "table_name": "project.dataset.table",
        "region": "US", 
        "limit": 1000
    }
).to_pandas()  # Returns pandas DataFrame

# Write DataFrame to BigQuery
result = zclient.write(
    df=df,
    full_table_path="project.dataset.new_table",
    write_type="truncate",  # or "append"
    confirm=True,           # Required for truncate - replaces all existing rows
    create_if_needed=True,  # Create table if it doesn't exist
    timeout=300            # Custom timeout (optional)
)

# Write operation examples
# Append to existing table (safe, no data loss)
result = zclient.write(df, "project.dataset.table", write_type="append")

# Truncate table (replaces all data). Without confirm=True this raises
# ZbqOperationError before anything is uploaded.
result = zclient.write(
    df,
    "project.dataset.table",
    write_type="truncate",
    confirm=True
)

# write_type must be "append" or "truncate" - anything else raises ValueError

# Create table if it doesn't exist
result = zclient.write(
    df,
    "project.dataset.new_table",
    create_if_needed=True   # Creates table with DataFrame schema
)

# Fail if table doesn't exist (strict mode)
result = zclient.write(
    df,
    "project.dataset.existing_table", 
    create_if_needed=False  # Raises error if table doesn't exist
)

# Statements that modify data go through execute()
result = zclient.execute(
    "UPDATE @table_name SET col = @new_value WHERE id = @target_id",
    parameters={"table_name": "project.dataset.table", "new_value": "updated", "target_id": 1}
)
print(result.num_dml_affected_rows)  # rows changed
print(result.job_id)

# insert()/update()/delete() are aliases for execute() - they don't check
# that the statement matches the method name
zclient.insert(
    "INSERT INTO @table_name (col1, col2) VALUES (@val1, @val2)",
    parameters={"table_name": "project.dataset.table", "val1": "data", "val2": 123}
)

# Multi-statement scripts (MERGE, DDL, CALL, etc.)
zclient.execute("""
    CREATE TEMP TABLE staged AS
    SELECT * FROM `project.dataset.incoming` WHERE region = @region;

    MERGE `project.dataset.target` T
    USING staged S ON T.id = S.id
    WHEN MATCHED THEN UPDATE SET T.value = S.value
    WHEN NOT MATCHED THEN INSERT (id, value) VALUES (S.id, S.value);
""", parameters={"region": "US"})  # Returns a JobResult

# Opt in to getting results back as a DataFrame (e.g. a script ending in SELECT)
df = zclient.execute("CALL `project.dataset.my_procedure`()", return_dataframe=True)

# Context manager support - automatic cleanup
with zclient as client:
    # Check the query and its cost first
    client.dry_run(
        "SELECT * FROM @table WHERE date >= @start_date",
        parameters={"table": "project.dataset.table1", "start_date": "2024-01-01"}
    )

    # Execute actual queries
    df1 = client.read(
        "SELECT * FROM @table WHERE date >= @start_date",
        parameters={"table": "project.dataset.table1", "start_date": "2024-01-01"}
    )
    df2 = client.read("SELECT * FROM table2")
    result = client.write(df1, "project.dataset.output_table")
# Client automatically cleaned up after context

Parameter Substitution & Advanced Features

Secure Parameter Handling

zbq uses BigQuery's native parameterized queries for complete SQL injection protection:

from zbq import zclient

# Basic secure parameterization
df = zclient.read(
    """SELECT * FROM `project.dataset.table` 
    WHERE region = @region AND status = @status""",
    parameters={
        "region": "US",           # Securely parameterized as STRING
        "status": "active"        # Securely parameterized as STRING  
    }
)

# Multiple data types with automatic type detection
df = zclient.read(
    """SELECT * FROM hotels 
    WHERE hotel_id = @id 
    AND price > @min_price 
    AND is_available = @available
    AND last_updated > @date""",
    parameters={
        "id": 123,                    # BigQuery INT64 parameter
        "min_price": 99.99,          # BigQuery FLOAT64 parameter
        "available": True,            # BigQuery BOOL parameter
        "date": None                 # BigQuery STRING parameter (NULL)
    }
)

# Table identifiers (handled separately for security)
df = zclient.read(
    "SELECT * FROM @table WHERE region = @region",
    parameters={
        "table": "project.dataset.hotels",  # Validated and backtick-quoted
        "region": "US"                      # Securely parameterized
    }
)

Array Parameters for Lists and IN Clauses

Use BigQuery's native ArrayQueryParameter for secure list handling:

# IN clause with string array
df = zclient.read(
    "SELECT * FROM hotels WHERE hotel_code IN UNNEST(@codes)",
    parameters={
        "codes": ['ABC', 'DEF', 'GHI']  # BigQuery ARRAY<STRING> parameter
    }
)

# Numeric array for IN clauses
df = zclient.read(
    "SELECT * FROM bookings WHERE booking_id IN UNNEST(@ids)", 
    parameters={
        "ids": [123, 456, 789]  # BigQuery ARRAY<INT64> parameter
    }
)

# Boolean arrays
df = zclient.read(
    "SELECT * FROM table WHERE flag IN UNNEST(@flags)",
    parameters={
        "flags": [True, False, True]  # BigQuery ARRAY<BOOL> parameter
    }
)

# Empty arrays (handled safely)
df = zclient.read(
    "SELECT * FROM table WHERE id IN UNNEST(@empty_ids)",
    parameters={
        "empty_ids": []  # BigQuery ARRAY<STRING> parameter (empty)
    }
)

Security Features

zbq provides complete protection against SQL injection attacks:

# All user inputs are safely parameterized
user_input = "'; DROP TABLE users; --"  # Malicious input
df = zclient.read(
    "SELECT * FROM users WHERE name = @user_name",
    parameters={
        "user_name": user_input  # Safely handled as STRING parameter
    }
)
# The malicious input is treated as a literal string value, not SQL code

# Table identifiers are validated separately
safe_table = "project.dataset.users"
df = zclient.read(
    "SELECT * FROM @table WHERE active = @status",
    parameters={
        "table": safe_table,    # Validated and safely quoted
        "status": True          # Securely parameterized as BOOL
    }
)

Read-Only Guard

read() is always read-only, and the check can't be turned off. Before running a query, zbq dry-runs it - free, no bytes processed - and refuses to run it unless BigQuery classifies it as a SELECT. Use execute() for statements that modify data.

zclient.read("SELECT * FROM `project.dataset.sales`")          # runs
zclient.read("DELETE FROM `project.dataset.sales` WHERE 1=1")  # raises ZbqOperationError
zclient.execute("DELETE FROM `project.dataset.sales` WHERE 1=1")  # runs

Because the check uses BigQuery's own parse rather than scanning the text:

• CTEs (WITH ... SELECT), subqueries, and INFORMATION_SCHEMA queries are allowed
• DML (INSERT, UPDATE, DELETE, MERGE), DDL, CALL, and EXPORT DATA are refused, including DML that uses a CTE
• Multi-statement scripts (anything with DECLARE, SET, or several ;-separated statements) are refused even if they only read; run those with execute(..., return_dataframe=True)
• Invalid SQL fails at the check with the same error the real run would give

The check costs one extra round trip per call, usually a few hundred milliseconds. For the strongest protection on untrusted SQL (user input, an LLM, an MCP tool call), also use credentials that only have BigQuery Data Viewer and Job User roles, so BigQuery itself rejects writes.

Dry Run

dry_run() previews your final query with parameters substituted into literal SQL, then validates it against BigQuery and reports how many bytes it would process - without running it or being billed. It works for any statement, including ones you'd pass to execute():

# Preview a query
info = zclient.dry_run(
    """SELECT * FROM @table
    WHERE region = @region
    AND hotel_codes IN UNNEST(@codes)
    AND rating >= @min_rating""",
    parameters={
        "table": "project.dataset.hotels",
        "region": "US",
        "codes": ['ABC', 'DEF'],
        "min_rating": 4
    }
)  # Logs the substituted query and byte estimate, returns a DryRunResult

# Output:
# DRY RUN - Query that would be executed:
# --------------------------------------------------
# SELECT * FROM `project.dataset.hotels`
# WHERE region = 'US' 
# AND hotel_codes IN UNNEST(['ABC', 'DEF'])
# AND rating >= 4
# --------------------------------------------------
# Parameters substituted above:
#   @region (STRING): 'US'
#   @codes (ARRAY<STRING>): ['ABC', 'DEF']
#   @min_rating (INT64): 4
# --------------------------------------------------
# Bytes that would be processed: 1.23 MB
# (Actual cost depends on your pricing model/edition and region.)
# --------------------------------------------------

If the query is invalid (bad column name, missing table, etc.), the substituted query still prints first, followed by BigQuery's validation error, which is then raised:

DRY RUN - Query that would be executed:
--------------------------------------------------
SELECT * FROM `project.dataset.hotels` WHERE region_typo = 'US'
--------------------------------------------------
DRY RUN VALIDATION FAILED: Unrecognized name: region_typo

The returned DryRunResult holds the same information for scripted checks (e.g. failing a job if a query would scan more than some byte threshold) instead of eyeballing the logged output:

from zbq import zclient

info = zclient.dry_run(
    "SELECT * FROM @table WHERE region = @region",
    parameters={"table": "project.dataset.hotels", "region": "US"},
)

if info.total_bytes_processed > 10 * 1024**3:  # 10 GB
    raise RuntimeError(f"Query would scan too much data: {info.total_bytes_processed:,} bytes")

print(info.statement_type)       # "SELECT"
print(info.referenced_tables)    # ["project.dataset.hotels"]

Pandas Integration

While zbq uses Polars internally for optimal performance, you can easily work with pandas DataFrames:

import pandas as pd
from zbq import zclient

# Get pandas DataFrame directly
df = zclient.read("SELECT * FROM `project.dataset.table`").to_pandas()
print(type(df))  # <class 'pandas.core.frame.DataFrame'>

# Works with all zbq features
pandas_df = zclient.read(
    """SELECT * FROM hotels 
    WHERE region = @region 
    AND hotel_codes IN UNNEST(@codes)
    AND rating >= @min_rating""",
    parameters={
        "region": "US",
        "codes": ['ABC', 'DEF'],
        "min_rating": 4
    }
).to_pandas()

# Mixed workflows - leverage both libraries' strengths
polars_df = zclient.read("SELECT * FROM large_table")  # Fast Polars processing

# Use Polars for heavy data transformation
processed_df = polars_df.filter(pl.col("amount") > 1000).group_by("category").sum()

# Convert to pandas for specific analysis or visualization
pandas_df = processed_df.to_pandas()

# Use pandas for plotting, ML libraries, etc.
import matplotlib.pyplot as plt
pandas_df.plot(kind='bar')
plt.show()

# Write back to BigQuery (Polars DataFrames only)
# Convert back if needed: pl.from_pandas(pandas_df)
zclient.write(polars_df, "project.dataset.analysis_results")

Why This Approach Works Well

• Performance: Polars handles the heavy lifting (SQL execution, data processing)
• Compatibility: Convert to pandas only when needed for specific libraries
• Flexibility: Use the best tool for each step in your workflow
• Memory efficiency: Polars' lazy evaluation optimizes the query execution

Timeout Configuration

Control query execution timeouts for long-running operations: The timeout covers the whole call - waiting for the query job and fetching its rows share one budget.

from zbq import zclient, BigQueryHandler

# Using default timeout (300 seconds / 5 minutes)
df = zclient.read("SELECT * FROM large_table")

# Custom timeout for specific query (10 minutes)
df = zclient.read(
    "SELECT * FROM very_large_table WHERE complex_calculation = @param",
    parameters={"param": "value"},
    timeout=600  # 10 minutes
)

# Set default timeout for all operations in a session
custom_client = BigQueryHandler(
    project_id="my-project",
    default_timeout=1200  # 20 minutes default
)

# All operations with this client use 20-minute timeout
df = custom_client.read("SELECT * FROM massive_dataset")

# Override default with specific timeout
df = custom_client.read("SELECT * FROM quick_query", timeout=30)  # 30 seconds

# Timeout works with all operations
custom_client.execute(
    "INSERT INTO @table VALUES (@val1, @val2)",
    parameters={"table": "my_table", "val1": "data", "val2": 123},
    timeout=300
)

# Handle timeout errors
try:
    df = zclient.read("SELECT * FROM enormous_table", timeout=60)
except TimeoutError as e:
    print(f"Query timed out: {e}")
    # Maybe try with longer timeout or different approach

Storage Operations

Single File Upload/Download

from zbq import zstorage

# Upload a single file directly (NEW!)
result = zstorage.upload_file(
    local_file_path="./report.xlsx",
    bucket_path="my-bucket",                # Uploads to bucket root as "report.xlsx"
)

# Upload single file to specific folder with custom name
result = zstorage.upload_file(
    local_file_path="./data/sales_2024.xlsx",
    bucket_path="my-bucket/reports/2024",   # Uploads to "reports/2024/" folder
    blob_name="january_sales.xlsx"          # Custom name in bucket
)

# Enhanced upload() method now handles both files AND directories
result = zstorage.upload(
    local_dir="./report.xlsx",              # File path (not directory)
    bucket_path="my-bucket/reports"         # Automatically detects it's a file
)
print(f"Single file upload: {result.uploaded_files} file, {result.total_bytes:,} bytes")

# Upload single file with error handling
try:
    result = zstorage.upload_file("./important_data.csv", "my-bucket/data")
    if result.uploaded_files == 1:
        print("File uploaded successfully!")
    else:
        print(f"Upload failed: {result.errors}")
except Exception as e:
    print(f"Upload error: {e}")

Basic Directory Upload/Download

from zbq import zstorage

# Simple directory upload with pattern - files go to bucket root
result = zstorage.upload(
    local_dir="./data",
    bucket_path="my-bucket",
    include_patterns="*.xlsx"  # Upload only Excel files
)

# Upload to specific folder in bucket
result = zstorage.upload(
    local_dir="./data", 
    bucket_path="my-bucket/reports/2024",  # Upload to reports/2024/ folder
    include_patterns="*.xlsx"
)

print(f"Uploaded {result.uploaded_files}/{result.total_files} files")
print(f"Total size: {result.total_bytes:,} bytes in {result.duration:.2f}s")

# Context manager support for batch operations
with zstorage as storage:
    # Mix single file and directory uploads
    result1 = storage.upload_file("./summary.pdf", "my-bucket/reports")
    result2 = storage.upload("./data1", "my-bucket/folder1", include_patterns="*.csv")
    result3 = storage.upload("./data2", "my-bucket/folder2", include_patterns="*.json")
    result4 = storage.download("my-bucket/archive", "./downloads", include_patterns="*.parquet")
# Storage client automatically cleaned up

# Simple download from bucket root
result = zstorage.download(
    bucket_path="my-bucket", 
    local_dir="./downloads",
    include_patterns="*.csv"  # Download only CSV files
)

# Download from specific folder in bucket
result = zstorage.download(
    bucket_path="my-bucket/data/exports",  # Download from data/exports/ folder
    local_dir="./downloads",
    include_patterns="*.csv"
)

Advanced Pattern Matching

# Multiple include patterns
result = zstorage.upload(
    local_dir="./reports",
    bucket_path="my-bucket/reports", 
    include_patterns=["*.xlsx", "*.csv", "*.json"],  # Multiple file types
    exclude_patterns=["temp_*", "*_backup.*"],       # Exclude temporary/backup files
    case_sensitive=False  # Case-insensitive matching
)

# Regex patterns for complex matching
result = zstorage.upload(
    local_dir="./logs",
    bucket_path="my-bucket/logs",
    include_patterns=r"log_\d{4}-\d{2}-\d{2}\.txt",  # Match log_YYYY-MM-DD.txt
    use_regex=True
)

Parallel Processing & Progress Tracking

# Automatic progress bar (shows for multiple files)
result = zstorage.upload(
    local_dir="./large-dataset", 
    bucket_path="my-bucket",
    include_patterns="*.xlsx",
    parallel=True,                    # Enable parallel uploads
    max_retries=5                     # Retry failed uploads
)
# Shows: "Uploading: 75%|███████▌  | 15/20 [00:30<00:10, 0.5files/s]"

# Custom progress callback (optional)
def progress_callback(completed, total):
    percentage = (completed / total) * 100
    print(f"Custom progress: {completed}/{total} files ({percentage:.1f}%)")

result = zstorage.upload(
    local_dir="./large-dataset", 
    bucket_path="my-bucket",
    progress_callback=progress_callback,
    show_progress=False               # Disable built-in progress bar
)

# Handle results
if result.failed_files > 0:
    print(f"WARNING: {result.failed_files} files failed to upload:")
    for error in result.errors:
        print(f"  - {error}")

print(f"Successfully uploaded {result.uploaded_files} files")
print(f"Total: {result.total_bytes:,} bytes in {result.duration:.2f}s")

Dry Run & Preview

# Preview what would be uploaded without actually uploading
result = zstorage.upload(
    local_dir="./data",
    bucket_path="my-bucket", 
    include_patterns="*.parquet",
    dry_run=True  # Preview only
)

print(f"Would upload {result.total_files} files ({result.total_bytes:,} bytes)")

# Progress bar control
result = zstorage.upload(
    local_dir="./data",
    bucket_path="my-bucket",
    include_patterns="*.xlsx", 
    show_progress=True     # Force show progress bar even for single files
)

result = zstorage.upload(
    local_dir="./data",
    bucket_path="my-bucket", 
    include_patterns="*.xlsx",
    show_progress=False    # Never show progress bar
)

Advanced Download with Filtering

# Download with path filtering and patterns
result = zstorage.download(
    bucket_path="my-data-bucket/reports/2024",  # Only files under this path
    local_dir="./downloaded-reports",
    include_patterns=["*.xlsx", "*.pdf"],
    exclude_patterns="*_draft.*",     # Skip draft files
    parallel=True,
    max_blobs_to_scan=500             # Scan at most 500 blobs before filtering
)

Reading Files Directly to Polars DataFrame

Basic Read - Auto-Concatenate Files

from zbq import zstorage

# Read all CSV files from a bucket folder into one DataFrame
df = zstorage.read("my-bucket/data", include_patterns="*.csv")
print(f"Loaded {len(df)} rows from GCS")

# Read all Parquet files
df = zstorage.read("my-bucket/parquet-data", include_patterns="*.parquet")

# Read multiple file types
df = zstorage.read(
    "my-bucket/data",
    include_patterns=["*.csv", "*.json"]
)

# Track which file each row came from
df = zstorage.read(
    "my-bucket/data",
    include_patterns="*.csv",
    add_source_column=True  # Adds "_source_file" column
)
print(df.select(["_source_file", "column1", "column2"]))

# Pass custom Polars reader options
df = zstorage.read(
    "my-bucket/data",
    include_patterns="*.csv",
    separator=";",           # Custom CSV separator
    has_header=True,
    infer_schema_length=1000
)

# Exclude certain patterns
df = zstorage.read(
    "my-bucket/data",
    include_patterns="*.csv",
    exclude_patterns=["temp_*", "*_backup.csv"]
)

# Specify file format explicitly
df = zstorage.read(
    "my-bucket/data",
    include_patterns="*.txt",
    file_format="csv"        # Treat .txt files as CSV
)

# Control parallelism and progress
df = zstorage.read(
    "my-bucket/data",
    include_patterns="*.csv",
    parallel=True,           # Read files in parallel (default)
    show_progress=True       # Show progress bar
)

Iterative Read - Process Files Individually

from zbq import zstorage
import polars as pl

# Collect DataFrames into a list
dataframes = []
for filename, df in zstorage.read_iter("my-bucket/data", include_patterns="*.csv"):
    dataframes.append(df)
    print(f"Read {filename}: {len(df)} rows")

# Process each file individually
for filename, df in zstorage.read_iter("my-bucket/data", include_patterns="*.csv"):
    print(f"{filename}: {len(df)} rows, columns: {df.columns}")
    # Custom processing per file
    if len(df) > 1000:
        print(f"  Large file detected!")

# Build a dictionary mapping filename -> DataFrame
dfs_dict = {
    name: df
    for name, df in zstorage.read_iter("my-bucket/data", include_patterns="*.csv")
}
print(f"Loaded {len(dfs_dict)} files")

# Conditional filtering - only keep large files
large_files = []
for filename, df in zstorage.read_iter("my-bucket/data", include_patterns="*.csv"):
    if len(df) > 1000:  # Only keep files with >1000 rows
        large_files.append((filename, df))

# Custom concatenation with source tracking
dataframes_with_source = []
for filename, df in zstorage.read_iter("my-bucket/data", include_patterns="*.csv"):
    df = df.with_columns(pl.lit(filename).alias("source"))
    dataframes_with_source.append(df)
combined = pl.concat(dataframes_with_source)

# Process different file types differently
for filename, df in zstorage.read_iter(
    "my-bucket/mixed-data",
    include_patterns=["*.csv", "*.parquet"]
):
    if filename.endswith('.csv'):
        # Special processing for CSV files
        df = df.with_columns(pl.col("date").str.strptime(pl.Date, "%Y-%m-%d"))
    else:
        # Parquet files are already typed correctly
        pass

    # Process df...

Supported File Formats

• CSV (.csv) - Supports all Polars CSV options (separator, has_header, etc.)
• Parquet (.parquet, .pq) - High performance columnar format
• JSON (.json) - Standard JSON files
• NDJSON (.jsonl, .ndjson) - Newline-delimited JSON

Schema Handling

# Files with different schemas are handled automatically
# Using pl.concat(how="diagonal") - fills missing columns with null

# file1.csv: [col_a, col_b]
# file2.csv: [col_a, col_c]
# Result df: [col_a, col_b, col_c] with nulls where columns don't exist

df = zstorage.read("my-bucket/data", include_patterns="*.csv")
# Automatically handles schema differences!

Practical Examples

# Example 1: Daily data aggregation
daily_sales = zstorage.read(
    "my-bucket/sales/2024-01",
    include_patterns="sales_2024-01-*.csv",
    add_source_column=True
)
# Process all January 2024 sales in one DataFrame

# Example 2: Multi-format data lake
df = zstorage.read(
    "my-bucket/data-lake/orders",
    include_patterns=["*.parquet", "*.csv"],
    file_format="auto"  # Auto-detect format
)

# Example 3: Conditional processing with read_iter
high_value_orders = []
for filename, df in zstorage.read_iter("my-bucket/orders", include_patterns="*.csv"):
    high_value = df.filter(pl.col("order_value") > 10000)
    if len(high_value) > 0:
        high_value_orders.append(high_value)

if high_value_orders:
    all_high_value = pl.concat(high_value_orders)
    print(f"Found {len(all_high_value)} high-value orders")

# Example 4: Load to BigQuery from GCS
from zbq import zclient, zstorage

# Read from GCS
df = zstorage.read("my-bucket/staging", include_patterns="*.parquet")

# Transform with Polars
transformed_df = (
    df
    .filter(pl.col("status") == "active")
    .select(["customer_id", "order_date", "amount"])
    .with_columns(pl.col("order_date").cast(pl.Date))
)

# Write to BigQuery
zclient.write(transformed_df, "project.dataset.orders", write_type="append")

Async & Concurrency

Every BigQueryHandler method blocks the thread that calls it until BigQuery responds. That is fine for scripts, but inside an asyncio application (an MCP server, FastAPI, etc.) a single blocking call freezes the whole event loop.

azclient is the async counterpart of zclient: the same methods as coroutines. Google's BigQuery client has no async API, so each call runs on a thread pool owned by the handler; the client releases the GIL while waiting on the network, so calls issued together genuinely overlap and the event loop stays free.

import asyncio
from zbq import azclient as azc

async def main():
    # Run several queries at once
    sales, users = await asyncio.gather(
        azc.read("SELECT * FROM `project.dataset.sales`"),
        azc.read("SELECT * FROM `project.dataset.users`"),
    )

    # Every zclient method is available, with the same arguments
    await azc.execute("CALL `project.dataset.refresh`()")
    sample = await azc.preview("project.dataset.sales")

    # catalog() lists datasets and tables concurrently
    tables = await azc.catalog()

    # Truncate writes require confirm=True, as with zclient
    await azc.write(sales, "project.dataset.sales_copy",
                    write_type="truncate", confirm=True)

asyncio.run(main())

azclient wraps zclient, so they share one project, timeout, credentials, and settings - setting zclient.project_id changes both. For different settings, create your own AsyncBigQueryHandler:

from zbq import AsyncBigQueryHandler

async with AsyncBigQueryHandler(project_id="other-project", max_concurrency=16) as aclient:
    df = await aclient.read("SELECT 1")

Notes:

• max_concurrency (default 8) caps how many BigQuery calls run at once; additional calls wait for a free worker.
• Wrap any existing handler to share its project and settings: AsyncBigQueryHandler(handler=my_handler).
• Cancelling a call (including with asyncio.wait_for) cancels the BigQuery jobs it submitted, so they stop running and billing.
• Changing project_id affects calls started afterwards; calls already in flight keep the project they started with. For concurrent work across projects, use one handler per project.
• A handler can be reused across separate asyncio.run() calls. async with or await aclient.aclose() shuts its thread pool down; the next call starts a fresh one, so the handler stays usable.
• Plain threads work too: BigQueryHandler methods are safe to call from a ThreadPoolExecutor without asyncio.

Scheduled Queries

ScheduledQueryHandler creates, inspects, runs and deletes BigQuery scheduled queries. They are managed by the BigQuery Data Transfer Service, which is an optional extra:

uv add "zbq[scheduling]"

You also need the BigQuery Data Transfer API enabled on the project, and permission to manage transfer configs.

from zbq import zscheduler

# Review what is scheduled, and how recent runs went
zscheduler.list_scheduled_queries()
zscheduler.scheduled_query_runs("abc123", limit=10)

# Create one (confirm=True is required: it runs and bills until deleted)
zscheduler.schedule_query(
    display_name="daily revenue",
    query="select current_date() as day, sum(amount) as revenue from `p.sales.orders`",
    schedule="every 24 hours",
    destination_table="reporting.daily_revenue",
    write_type="append",
    confirm=True,
)

# Pause, resume, change, run now, delete
zscheduler.update_scheduled_query("abc123", disabled=True)
zscheduler.update_scheduled_query("abc123", schedule="every mon 09:00")
zscheduler.run_scheduled_query("abc123")
zscheduler.delete_scheduled_query("abc123", confirm=True)

zscheduler is a ready-made ScheduledQueryHandler for the "us" location, like zclient and zstorage. The location must match the destination dataset's location, so for another one create your own handler:

from zbq import ScheduledQueryHandler

eu_scheduler = ScheduledQueryHandler(location="eu")

Notes:

• Every method that takes name needs the full resource name (projects/.../transferConfigs/ID) or the bare config ID. It does not accept the display name: display names are not unique, and a name with spaces is rejected with ValueError. Get the identifier from the name column of list_scheduled_queries(), and use the last path segment as the bare ID.
• list_scheduled_queries() returns these columns: name, display_name, schedule, disabled, state, destination_dataset, destination_table, write_type, query, next_run_time, update_time, owner_email. It is a summary; other settings of the transfer config (such as query parameters, partitioning or notification topic) are not included. owner_email is null when the API does not report an owner.
• destination_table is for a SELECT's results. Leave it out for DML and DDL. It must be in the handler's project.
• The query is not dry-run before it is scheduled, because scheduled queries can use @run_time and @run_date, which a dry run cannot resolve. Preview with zclient.dry_run(query, parameters=...).
• Plain user credentials may be refused when creating or changing a schedule. Pass service_account="name@project.iam.gserviceaccount.com" to schedule_query() to run it as a service account.
• run_scheduled_query(name, run_time=...) is also how you backfill a past date. The run executes and bills like any other.
• A disabled (paused) query cannot be run manually. Resume it first with update_scheduled_query(name, disabled=False), run it, then pause it again.
• delete_scheduled_query() is permanent and removes the run history. Use update_scheduled_query(name, disabled=True) to stop one temporarily.
• This handler is synchronous only.

Advanced Configuration

Project Configuration

zbq automatically detects your Google Cloud project but also provides flexible project management:

from zbq import zclient, zstorage

# Check current project (auto-detected)
print(f"Current project: {zclient.project_id}")

# Manual project override
zclient.project_id = "my-specific-project"
zstorage.project_id = "my-storage-project"  # Can be different

# Verify the change
print(f"BigQuery project: {zclient.project_id}")
print(f"Storage project: {zstorage.project_id}")

# Project-specific operations
df = zclient.read("SELECT * FROM dataset.table")  # Uses my-specific-project

Project Detection Order

zbq uses the first of these that is set:

1. Constructor parameter: BigQueryHandler(project_id="project"), or zclient.project_id = "project" on an existing handler
2. Environment variable: GOOGLE_CLOUD_PROJECT
3. Environment variable: CLOUDSDK_CORE_PROJECT (gcloud's own override)
4. Active gcloud configuration: the same project gcloud config get-value project shows, including after gcloud config configurations activate <name>

Environment variables win over gcloud config, matching Google's client libraries, so a CI job or container can override the machine's setting. CLOUDSDK_CONFIG and CLOUDSDK_ACTIVE_CONFIG_NAME are honored when locating the gcloud configuration. If nothing is set, the first BigQuery or Storage call raises ZbqConfigurationError with setup instructions.

# Different ways to set project
from zbq import BigQueryHandler, StorageHandler

# Method 1: Constructor (recommended for apps)
bq = BigQueryHandler(project_id="my-project")

# Method 2: Property setter (good for interactive use)
bq = BigQueryHandler()  # Auto-detects project
bq.project_id = "different-project"  # Override if needed

# Method 3: Environment variable
# Set GOOGLE_CLOUD_PROJECT before starting Python. zclient, azclient, and
# zstorage detect their project when zbq is imported, so setting it from
# Python afterwards only affects handlers you create later.
bq = BigQueryHandler()  # Uses the environment variable

# Method 4: gcloud config (system default)
# Run: gcloud config set project my-default-project
bq = BigQueryHandler()  # Uses my-default-project

$env.GOOGLE_CLOUD_PROJECT = "env-project"
uv run python my_script.py

Multi-Project Workflows

# Work with multiple projects simultaneously
prod_client = BigQueryHandler(project_id="prod-project")
dev_client = BigQueryHandler(project_id="dev-project")

# Different data sources
prod_df = prod_client.read("SELECT * FROM prod_dataset.table")
dev_df = dev_client.read("SELECT * FROM dev_dataset.table")

# Cross-project operations
result = prod_client.write(dev_df, "prod_dataset.migrated_table")

Custom Logging

zbq logs through loguru but never adds or removes sinks, so its messages go wherever your application's loguru configuration sends them. With loguru's default setup that is stderr, so zbq output (including the dry_run() preview) is visible without any configuration.

from loguru import logger

# Silence zbq entirely
logger.disable("zbq")

# Turn it back on
logger.enable("zbq")

# Send only zbq records at INFO and above to their own file
logger.add("logs/zbq.log", level="INFO", filter="zbq")

# Keep zbq records out of an application sink
import sys
logger.add(sys.stderr, filter=lambda record: not record["name"].startswith("zbq"))

Error Handling

from zbq import ZbqAuthenticationError, ZbqOperationError, ZbqConfigurationError

# BigQuery error handling
try:
    df = zclient.read(
        "SELECT * FROM @table WHERE column = @value",
        parameters={"table": "project.dataset.table", "value": "test"}
    )
except ZbqAuthenticationError:
    print("Authentication failed. Run: gcloud auth application-default login")
except ZbqConfigurationError:
    print("Configuration error. Check your project settings.")
except ZbqOperationError as e:
    print(f"BigQuery operation failed: {e}")
except TimeoutError as e:
    print(f"Query timed out: {e}")
    print("Consider increasing timeout or optimizing the query")

# Write operation errors
try:
    zclient.write(
        empty_df, 
        "project.dataset.table",
        write_type="truncate",
        confirm=True
    )
except ValueError as e:
    if "Missing required argument" in str(e):
        print("DataFrame is empty or table path is missing")
    else:
        print(f"Write validation error: {e}")  # e.g. unknown write_type
except ZbqOperationError as e:
    print(f"Write operation failed: {e}")  # e.g. truncate without confirm=True

# Storage operations error handling
try:
    result = zstorage.upload("./data", "my-bucket", include_patterns="*.csv")
except ZbqAuthenticationError:
    print("Authentication failed. Run: gcloud auth application-default login")
except ZbqConfigurationError:
    print("Configuration error. Check your project settings.")
except ZbqOperationError as e:
    print(f"Storage operation failed: {e}")

# Comprehensive error handling for production use
def safe_bigquery_operation():
    try:
        # Your BigQuery operations here
        df = zclient.read("SELECT * FROM large_table", timeout=300)
        return df
    
    except ZbqAuthenticationError:
        print("Authentication Error")
        print("Run: gcloud auth application-default login")
        return None
        
    except ZbqConfigurationError as e:
        print("Configuration Error")
        print(f"Check project settings: {e}")
        return None
        
    except TimeoutError:
        print("Query Timeout")
        print("Try increasing timeout or optimizing query")
        return None
        
    except ZbqOperationError as e:
        print("BigQuery Operation Error") 
        print(f"Details: {e}")
        return None
        
    except Exception as e:
        print(f"Unexpected error: {type(e).__name__}: {e}")
        return None

Working with Results

from zbq import UploadResult, DownloadResult

# Upload with detailed result handling
result: UploadResult = zstorage.upload(
    local_dir="./data",
    bucket_path="my-bucket",
    include_patterns=["*.json", "*.csv"]
)

# Detailed statistics
print(f"""
Upload Summary:
Total files: {result.total_files}
Uploaded: {result.uploaded_files} 
Skipped: {result.skipped_files}
Failed: {result.failed_files}
Total size: {result.total_bytes:,} bytes
Duration: {result.duration:.2f} seconds
""")

# Handle errors
if result.errors:
    print("Errors encountered:")
    for error in result.errors[:5]:  # Show first 5 errors
        print(f"  - {error}")
    if len(result.errors) > 5:
        print(f"  ... and {len(result.errors) - 5} more errors")

Pattern Matching Guide

Glob Patterns (Default)

• *.xlsx - All Excel files
• data_*.csv - CSV files starting with "data_"
• report_????_??.pdf - Reports with specific naming pattern
• [!.]*.txt - Text files not starting with dot

Patterns are matched against the file name only, not the directory path. A pattern such as **/*.json or .git/* therefore does not select by directory. Use bucket_path (for downloads and reads) or local_dir to choose the directory, and patterns to filter names within it.

Regex Patterns

# Enable regex with use_regex=True
zstorage.upload(
    local_dir="./logs",
    bucket_path="my-bucket",
    include_patterns=[
        r"access_log_\d{4}-\d{2}-\d{2}\.log",  # access_log_2024-01-01.log
        r"error_log_\d{8}\.log"                # error_log_20240101.log  
    ],
    use_regex=True
)

Complex Filtering

# Include multiple types, exclude temp files
zstorage.upload(
    local_dir="./workspace",
    bucket_path="my-bucket",
    include_patterns=["*.py", "*.json", "*.md", "*.yml"],
    exclude_patterns=["*.pyc", "temp_*"],
    case_sensitive=False
)

Authentication & Setup

1. Install Google Cloud SDK:

   gcloud auth application-default login
   gcloud config set project YOUR_PROJECT_ID
   

2. Or set environment variables:

   $env.GOOGLE_APPLICATION_CREDENTIALS = "path/to/service-account.json"
   $env.GOOGLE_CLOUD_PROJECT = "your-project-id"
   

Requirements

• Python ≥ 3.11
• Google Cloud project with BigQuery and/or Storage APIs enabled
• Appropriate IAM permissions for your operations

Performance Tips

BigQuery Operations

1. Use parameterized queries for better performance and security:

   # Good: Parameters are processed efficiently
   df = zclient.read("SELECT * FROM table WHERE id = @id", parameters={"id": 123})
   
   # Avoid: String concatenation
   df = zclient.read(f"SELECT * FROM table WHERE id = {user_id}")
   

2. Set appropriate timeouts for query complexity:

   # Quick queries: shorter timeout
   df = zclient.read("SELECT COUNT(*) FROM table", timeout=30)
   
   # Complex analytics: longer timeout
   df = zclient.read("SELECT * FROM huge_table", timeout=1800)  # 30 minutes
   

3. Optimize DataFrame conversions:

   # Efficient: Convert only when needed
   polars_df = zclient.read("SELECT * FROM table")
   processed_df = polars_df.filter(pl.col("amount") > 1000)  # Fast Polars operations
   pandas_df = processed_df.to_pandas()  # Convert at the end
   
   # Less efficient: Convert immediately
   pandas_df = zclient.read("SELECT * FROM table").to_pandas()
   # Now all operations use slower pandas
   

4. Use dry_run() for query optimization:

   # Preview complex queries before execution
   zclient.dry_run(complex_query, parameters=params)
   # Review the generated SQL and the bytes-processed estimate to spot
   # missing partition filters or overly broad scans before running it
   

5. Run independent queries concurrently with azclient (see Async & Concurrency):

   from zbq import azclient as azc
   
   df1, df2 = await asyncio.gather(
       azc.read("SELECT * FROM table1"),
       azc.read("SELECT * FROM table2"),
   )
   

   Credentials are resolved once per handler and shared by every call, so reuse a handler rather than creating new ones repeatedly.

6. Memory management for large datasets:

   # Process large datasets in chunks
   for i in range(0, total_records, chunk_size):
       chunk_df = zclient.read(
           "SELECT * FROM table LIMIT @limit OFFSET @offset",
           parameters={"limit": chunk_size, "offset": i}
       )
       # Process chunk...
   

Storage Operations

1. Use parallel processing for multiple files: parallel=True
2. Adjust thread count based on your system: max_workers=8
3. Use dry-run to preview large operations first
4. Filter early with specific patterns to avoid processing unwanted files
5. Monitor progress with callback functions for long operations
6. Use upload_file() for single files instead of directory patterns when possible

API Reference

BigQuery Methods

zclient.read(query, timeout=None, parameters=None, max_bytes_billed=None)

Execute a SELECT query and return results as a Polars DataFrame. Always read-only: anything BigQuery doesn't classify as a SELECT raises ZbqOperationError before it runs. See Read-Only Guard.

• query (str): SQL query with optional @param_name placeholders
• timeout (int, optional): Query timeout in seconds (default: 300)
• parameters (dict, optional): Parameter values for @param_name substitution
• max_bytes_billed (int | str, optional): Cost limit; see Cost Limit (default: None, no limit)
• Returns: pl.DataFrame

A timeout must be positive; zero or negative raises ValueError. A timeout raises the builtin TimeoutError, but the BigQuery job is not cancelled and may keep running; only an async cancellation cancels it.

zclient.dry_run(query, parameters=None)

Log the query with parameters substituted into literals, validate it against BigQuery, and report the bytes it would process, without executing it. Works for any statement type.

• query (str): SQL statement(s) with optional @param_name placeholders
• parameters (dict, optional): Parameter values for @param_name substitution
• Returns: DryRunResult

zclient.preview(full_table_path, max_results=10, timeout=None)

Fetch a sample of rows from a table via the tabledata.list API instead of a query job - scans no bytes and isn't billed. Use it to peek at a table's shape/content before writing a real query.

• full_table_path (str): Table identifier, e.g. "project.dataset.table"
• max_results (int, optional): Maximum number of rows to fetch (default: 10)
• timeout (int, optional): Timeout in seconds (default: 300)
• Returns: pl.DataFrame

zclient.execute(query, timeout=None, parameters=None, return_dataframe=False, max_bytes_billed=None)

Execute an arbitrary SQL statement or multi-statement script - DML, MERGE, DDL (CREATE/DROP/ALTER), CALL, or scripts that mix several operations together. Unlike read(), it is not restricted to SELECT.

• query (str): SQL statement(s) with optional @param_name placeholders
• timeout (int, optional): Query timeout in seconds (default: 300)
• parameters (dict, optional): Parameter values for @param_name substitution
• return_dataframe (bool, optional): Fetch and return results as a pl.DataFrame instead of a JobResult (default: False)
• max_bytes_billed (int | str, optional): Cost limit; see Cost Limit (default: None, no limit)
• Returns: JobResult by default, or pl.DataFrame if return_dataframe=True

zclient.insert/update/delete(query, timeout=None, parameters=None, max_bytes_billed=None)

Aliases for execute() that return a JobResult. They do not check that the statement matches the method name.

Cost Limit

read(), execute(), insert(), update() and delete() accept max_bytes_billed. It is passed to BigQuery as the job's maximumBytesBilled, so BigQuery fails the job without charge if it would bill more than the limit.

df = zclient.read(query, max_bytes_billed="100 GB")
zclient.execute("DELETE FROM `p.d.t` WHERE true", max_bytes_billed=50 * 1024**3)

• An int is a byte count. A string takes a unit (B, KB, MB, GB, TB, PB, any case, e.g. "1.5 TB").
• Units are binary, matching BigQuery billing: 1 GB is 2**30 bytes.
• None (default) sets no limit. Zero, negative and malformed values raise ValueError before any job is submitted.
• The limit applies to the query job only. The read-only guard's dry run is never billed, and dry_run() takes no limit.
• On-demand queries bill a 10 MB minimum per table referenced, so a very small limit can reject a query on a tiny table. The error comes from BigQuery, not zbq.

JobResult

Returned by execute(), insert(), update(), and delete() for statements that don't return rows.

• job_id (str | None): BigQuery job ID
• statement_type (str | None): BigQuery's classification, e.g. "INSERT", "MERGE", "SCRIPT"
• num_dml_affected_rows (int | None): Rows inserted, updated, or deleted; None for non-DML statements
• total_bytes_processed (int | None): Bytes the job processed

DryRunResult

Returned by dry_run(), for scripted checks (e.g. failing a CI job if a query would scan too many bytes) instead of eyeballing the logged preview.

• statement_type (str | None): BigQuery's classification, e.g. "SELECT", "INSERT"
• referenced_tables (List[str]): Tables the query touches, as "project.dataset.table"
• total_bytes_processed (int): Bytes the query would process if run
• query (str): The query with @param_name placeholders substituted for their literal values (the same text that is logged)

zclient.write(df, full_table_path, write_type="append", confirm=False, create_if_needed=True, timeout=None)

Write DataFrame to BigQuery table.

• df (pl.DataFrame): Data to write
• full_table_path (str): Complete table path "project.dataset.table"
• write_type ("append" | "truncate"): Append rows, or replace the table's contents (default: "append"). Any other value raises ValueError
• confirm (bool): Must be True for a truncate write, or it raises ZbqOperationError (default: False). Ignored for appends
• create_if_needed (bool): Create table if it doesn't exist (default: True)
• timeout (int, optional): Operation timeout in seconds
• Returns: The load job's final state, e.g. "DONE"

Parameter Types Support

• Strings: BigQuery STRING parameters (automatically quoted and escaped)
• Integers: BigQuery INT64 parameters
• Floats: BigQuery FLOAT64 parameters
• Booleans: BigQuery BOOL parameters
• None: BigQuery STRING parameters with NULL value (use TypedParam for other types)
• Lists/Tuples: BigQuery ARRAY<type> parameters (use with UNNEST()); the type comes from the first non-None element, and an empty list is ARRAY<STRING>
• Dates: BigQuery DATE parameters (date objects)
• Timestamps: BigQuery TIMESTAMP parameters (datetime objects)
• Other types: Anything else, such as Decimal, is sent as a STRING of its str() value (use TypedParam for NUMERIC and similar)
• Table identifiers: Separately validated and backtick-quoted (not parameterized)

Explicit Types with TypedParam

Wrap a value in TypedParam(value, bq_type) when inference picks the wrong type or cannot pick one. Plain values are unaffected.

from datetime import datetime
from decimal import Decimal
from zbq import TypedParam, zclient

df = zclient.read(
    "select * from `p.d.t` where @min_id is null or id >= @min_id",
    parameters={"min_id": TypedParam(None, "INT64")},      # typed NULL
)

zclient.read(
    "select * from `p.d.t` where id in unnest(@ids)",
    parameters={"ids": TypedParam([], "INT64")},           # empty list, typed array
)

zclient.read(
    "select * from `p.d.t` where created > @since and amount > @min_amount",
    parameters={
        "since": TypedParam(datetime(2024, 1, 1), "DATETIME"),   # not TIMESTAMP
        "min_amount": TypedParam(Decimal("10.50"), "NUMERIC"),
    },
)

• bq_type: One of BOOL, INT64, FLOAT64, NUMERIC, BIGNUMERIC, STRING, BYTES, DATE, DATETIME, TIME, TIMESTAMP (case-insensitive). Any other name raises ValueError.
• A list or tuple value becomes an ARRAY of bq_type; any other value is a scalar.
• TypedParam does not apply to the table and table_name parameters, which are substituted as identifiers.

Common Use Cases & Practical Examples

Data Pipeline: File Processing

from zbq import zclient, zstorage
import polars as pl

# Complete data pipeline example
def process_daily_files(date_str):
    """Process daily CSV files and load to BigQuery"""
    
    # 1. Upload today's files to staging
    upload_result = zstorage.upload(
        local_dir=f"./data/{date_str}",
        bucket_path="my-bucket/staging",
        include_patterns="*.csv"
    )
    
    print(f"Uploaded {upload_result.uploaded_files} files for {date_str}")
    
    # 2. Process and validate the data
    df = zclient.read(f"""
        SELECT 
            customer_id,
            order_date,
            amount,
            status
        FROM temp_external_table
        WHERE order_date = @date
        AND amount > 0
    """, parameters={"date": date_str})
    
    # 3. Data quality checks with Polars
    invalid_orders = df.filter(pl.col("customer_id").is_null()).height
    if invalid_orders > 0:
        print(f"Warning: {invalid_orders} orders with missing customer_id")
    
    # 4. Load clean data to production table
    clean_df = df.filter(pl.col("customer_id").is_not_null())
    
    result = zclient.write(
        df=clean_df,
        full_table_path="prod.orders.daily_orders",
        write_type="append",
        create_if_needed=True
    )
    
    print(f"Loaded {len(clean_df)} orders to BigQuery")
    
    return len(clean_df)

# Run pipeline
orders_processed = process_daily_files("2024-01-15")

Backup & Restore Operations

from zbq import zclient, zstorage
from datetime import datetime

def backup_table_to_storage(table_path, backup_bucket):
    """Export BigQuery table to Cloud Storage"""
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    # Export table data
    df = zclient.read(f"SELECT * FROM {table_path}")
    
    # Save locally first
    local_file = f"./backup_{timestamp}.parquet"
    df.write_parquet(local_file)
    
    # Upload to storage
    result = zstorage.upload_file(
        local_file_path=local_file,
        bucket_path=f"{backup_bucket}/backups",
        blob_name=f"{table_path.replace('.', '_')}_{timestamp}.parquet"
    )
    
    if result.uploaded_files == 1:
        print(f"Table {table_path} backed up successfully")
        # Clean up local file
        import os
        os.remove(local_file)
    else:
        print(f"Backup failed: {result.errors}")
    
    return result

def restore_table_from_backup(backup_file, target_table):
    """Restore table from backup file"""
    
    # Download backup file
    zstorage.download(
        bucket_path=f"my-bucket/backups/{backup_file}",
        local_dir="./restore",
        include_patterns=backup_file
    )
    
    # Load and restore
    df = pl.read_parquet(f"./restore/{backup_file}")
    
    result = zclient.write(
        df=df,
        full_table_path=target_table,
        write_type="truncate",
        confirm=True  # Explicitly confirm replacing the table
    )
    
    print(f"Restored {len(df)} records to {target_table}")

# Usage
backup_result = backup_table_to_storage("prod.sales.transactions", "backup-bucket")
# restore_table_from_backup("prod_sales_transactions_20240115_143022.parquet", "dev.sales.transactions")

Report Generation & Distribution

from zbq import zclient, zstorage
import polars as pl

def generate_monthly_report(year, month):
    """Generate monthly sales report and distribute"""
    
    # Generate comprehensive report data
    report_df = zclient.read("""
        SELECT 
            region,
            product_category,
            SUM(revenue) as total_revenue,
            COUNT(DISTINCT customer_id) as unique_customers,
            AVG(order_value) as avg_order_value,
            COUNT(*) as total_orders
        FROM sales_data 
        WHERE EXTRACT(YEAR FROM order_date) = @year 
        AND EXTRACT(MONTH FROM order_date) = @month
        GROUP BY region, product_category
        ORDER BY total_revenue DESC
    """, parameters={"year": year, "month": month})
    
    # Save report in multiple formats
    report_name = f"monthly_report_{year}_{month:02d}"
    
    # Save as Excel for business users
    pandas_df = report_df.to_pandas()
    excel_file = f"./reports/{report_name}.xlsx"
    pandas_df.to_excel(excel_file, index=False)
    
    # Save as Parquet for analysts
    parquet_file = f"./reports/{report_name}.parquet"
    report_df.write_parquet(parquet_file)
    
    # Upload both formats to shared storage
    excel_result = zstorage.upload_file(
        local_file_path=excel_file,
        bucket_path="company-reports/monthly",
        blob_name=f"{report_name}.xlsx"
    )
    
    parquet_result = zstorage.upload_file(
        local_file_path=parquet_file,
        bucket_path="company-reports/monthly", 
        blob_name=f"{report_name}.parquet"
    )
    
    # Store report metadata in BigQuery
    metadata_df = pl.DataFrame({
        "report_name": [report_name],
        "report_date": [f"{year}-{month:02d}-01"],
        "total_revenue": [report_df.select(pl.sum("total_revenue")).item()],
        "total_customers": [report_df.select(pl.sum("unique_customers")).item()],
        "generated_at": [datetime.now()],
        "excel_uploaded": [excel_result.uploaded_files == 1],
        "parquet_uploaded": [parquet_result.uploaded_files == 1]
    })
    
    zclient.write(
        df=metadata_df,
        full_table_path="analytics.reports.monthly_metadata", 
        write_type="append"
    )
    
    print(f"Generated report for {year}-{month:02d}")
    print(f"Excel: {'Success' if excel_result.uploaded_files == 1 else 'Failed'}")
    print(f"Parquet: {'Success' if parquet_result.uploaded_files == 1 else 'Failed'}")
    
    return report_df

# Generate reports for Q1 2024
for month in [1, 2, 3]:
    generate_monthly_report(2024, month)

Data Quality Monitoring

from zbq import zclient
import polars as pl

def monitor_data_quality(table_name):
    """Comprehensive data quality monitoring"""
    
    # Run multiple quality checks in parallel
    quality_checks = {}
    
    # 1. Basic statistics
    stats_df = zclient.read(
        """
        SELECT 
            COUNT(*) as total_rows,
            COUNT(DISTINCT customer_id) as unique_customers,
            MIN(order_date) as earliest_date,
            MAX(order_date) as latest_date,
            AVG(order_value) as avg_order_value,
            STDDEV(order_value) as stddev_order_value
        FROM @table_name
        WHERE DATE(order_date) = CURRENT_DATE()
        """,
        parameters={"table_name": table_name}
    )
    
    # 2. Data completeness
    completeness_df = zclient.read(
        """
        SELECT 
            'customer_id' as column_name,
            COUNTIF(customer_id IS NULL) as null_count,
            COUNT(*) as total_count,
            ROUND(COUNTIF(customer_id IS NOT NULL) / COUNT(*) * 100, 2) as completeness_pct
        FROM @table_name
        WHERE DATE(order_date) = CURRENT_DATE()
        
        UNION ALL
        
        SELECT 
            'order_value' as column_name,
            COUNTIF(order_value IS NULL) as null_count,
            COUNT(*) as total_count,
            ROUND(COUNTIF(order_value IS NOT NULL) / COUNT(*) * 100, 2) as completeness_pct
        FROM @table_name
        WHERE DATE(order_date) = CURRENT_DATE()
        """,
        parameters={"table_name": table_name}
    )
    
    # 3. Anomaly detection
    anomalies_df = zclient.read(
        """
        SELECT 
            customer_id,
            order_value,
            order_date,
            'high_value' as anomaly_type
        FROM @table_name
        WHERE DATE(order_date) = CURRENT_DATE()
        AND order_value > (
            SELECT AVG(order_value) + 3 * STDDEV(order_value) 
            FROM @table_name
            WHERE DATE(order_date) >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
        )
        
        UNION ALL
        
        SELECT 
            customer_id,
            order_value,
            order_date,
            'negative_value' as anomaly_type
        FROM @table_name
        WHERE DATE(order_date) = CURRENT_DATE()
        AND order_value < 0
        """,
        parameters={"table_name": table_name}
    )
    
    # Process results
    total_rows = stats_df.item(0, "total_rows")
    avg_completeness = completeness_df.select(pl.mean("completeness_pct")).item()
    anomaly_count = len(anomalies_df)
    
    # Quality score calculation
    quality_score = min(100, avg_completeness - (anomaly_count / total_rows * 100))
    
    # Store monitoring results
    monitoring_df = pl.DataFrame({
        "check_date": [datetime.now().date()],
        "table_name": [table_name],
        "total_rows": [total_rows],
        "avg_completeness": [avg_completeness],
        "anomaly_count": [anomaly_count],
        "quality_score": [quality_score],
        "passed": [quality_score >= 95]
    })
    
    zclient.write(
        df=monitoring_df,
        full_table_path="monitoring.data_quality.daily_checks",
        write_type="append"
    )
    
    # Alert if quality issues
    if quality_score < 95:
        print(f"Data quality alert for {table_name}")
        print(f"   Quality score: {quality_score:.1f}%")
        print(f"   Anomalies found: {anomaly_count}")
        print(f"   Avg completeness: {avg_completeness:.1f}%")
    else:
        print(f"Data quality check passed for {table_name}")
    
    return {
        "quality_score": quality_score,
        "total_rows": total_rows,
        "anomalies": anomaly_count
    }

# Monitor key tables
results = monitor_data_quality("prod.orders.daily_transactions")

Storage Methods

zstorage.upload_file(local_file_path, bucket_path, blob_name=None, max_retries=3)

Upload a single file to Google Cloud Storage.

• local_file_path (str | Path): Path to the local file to upload
• bucket_path (str): GCS bucket path (e.g., "my-bucket" or "my-bucket/folder/")
• blob_name (str, optional): Custom name for the blob (defaults to filename)
• max_retries (int): Number of retry attempts for failed upload (default: 3)
• Returns: UploadResult with detailed statistics

zstorage.upload(local_dir, bucket_path, include_patterns=None, exclude_patterns=None, **kwargs)

Upload files or a single file to Google Cloud Storage with pattern matching.

• local_dir (str | Path): Local directory OR single file path to upload
• bucket_path (str): GCS bucket path
• include_patterns (str | List[str], optional): Patterns to include (ignored for single files)
• exclude_patterns (str | List[str], optional): Patterns to exclude (ignored for single files)
• case_sensitive (bool): Whether pattern matching is case sensitive (default: True)
• use_regex (bool): Use regex patterns instead of glob patterns (default: False)
• dry_run (bool): Preview operation without executing (default: False)
• parallel (bool): Use parallel uploads for multiple files (default: True)
• max_retries (int): Number of retry attempts (default: 3)
• Returns: UploadResult with detailed statistics

zstorage.download(bucket_path, local_dir, include_patterns=None, **kwargs)

Download files from Google Cloud Storage with pattern matching.

• bucket_path (str): GCS bucket path to download from
• local_dir (str): Local directory to download files to
• include_patterns (str | List[str], optional): Patterns to include
• exclude_patterns (str | List[str], optional): Patterns to exclude
• parallel (bool): Use parallel downloads (default: True)
• max_blobs_to_scan (int, optional): Maximum number of blobs to scan before applying patterns (default: None, no limit)
• Returns: DownloadResult with detailed statistics

zstorage.read(bucket_path, include_patterns=None, exclude_patterns=None, **kwargs)

Read files from Google Cloud Storage into a Polars DataFrame.

• bucket_path (str): GCS bucket path to read from
• include_patterns (str | List[str], optional): Patterns to include (e.g., "*.csv")
• exclude_patterns (str | List[str], optional): Patterns to exclude
• case_sensitive (bool): Whether pattern matching is case sensitive (default: True)
• use_regex (bool): Use regex patterns instead of glob (default: False)
• file_format (str): File format - "auto", "csv", "parquet", "json", "ndjson" (default: "auto")
• parallel (bool): Use parallel reading (default: True)
• add_source_column (bool): Add "_source_file" column (default: False)
• show_progress (bool, optional): Show progress bar (auto-detect if None)
• max_retries (int): Number of retry attempts (default: 3)
• max_blobs_to_scan (int, optional): Maximum number of blobs to scan before applying patterns (default: None, no limit)
• reader_options: Additional options for Polars read functions (e.g., separator=";", has_header=True)
• Returns: pl.DataFrame with combined data from all matching files

zstorage.read_iter(bucket_path, include_patterns=None, **kwargs)

Iterator that yields individual DataFrames from GCS files. A file that cannot be read is logged and skipped by default, so the caller cannot tell it was missed; pass on_error="raise" to stop with a ZbqOperationError instead.

• bucket_path (str): GCS bucket path to read from
• include_patterns (str | List[str], optional): Patterns to include
• exclude_patterns (str | List[str], optional): Patterns to exclude
• file_format (str): File format - "auto", "csv", "parquet", "json", "ndjson" (default: "auto")
• max_retries (int): Number of retry attempts (default: 3)
• max_blobs_to_scan (int, optional): Maximum number of blobs to scan before applying patterns (default: None, no limit)
• on_error ("skip" | "raise"): What to do when a file cannot be read (default: "skip")
• reader_options: Additional options for Polars read functions
• Yields: Tuple[str, pl.DataFrame] - (filename, dataframe) for each file

Constructor Options

BigQueryHandler(project_id="", default_timeout=300)

Create a custom BigQuery client.

• project_id (str, optional): GCP project ID (auto-detected if empty)
• default_timeout (int): Default timeout for all operations in seconds (default: 300)

AsyncBigQueryHandler(project_id="", default_timeout=300, max_concurrency=8, handler=None)

Asyncio interface with the same methods as BigQueryHandler, as coroutines. zbq.azclient is a ready-made instance wrapping zclient.

• project_id, default_timeout: As for BigQueryHandler; ignored if handler is given
• max_concurrency (int): Maximum BigQuery calls in flight at once (default: 8)
• handler (BigQueryHandler, optional): Existing handler to wrap, e.g. zclient

StorageHandler(project_id="", max_workers=4)

Create a custom Storage client.

• project_id (str, optional): GCP project ID (auto-detected if empty)
• max_workers (int): Thread pool size for parallel operations (default: 4)

ScheduledQueryHandler(project_id="", location="us", default_timeout=60)

Manage BigQuery scheduled queries. Needs the scheduling extra; see Scheduled Queries. zbq.zscheduler is a ready-made instance with the defaults.

• project_id (str, optional): GCP project ID (auto-detected if empty)
• location (str): Location of the scheduled queries, matching the destination dataset (default: "us")
• default_timeout (int): Timeout in seconds for each API call (default: 60)
• Methods: list_scheduled_queries(), get_scheduled_query(name), scheduled_query_runs(name, limit=20), schedule_query(display_name, query, schedule, destination_table=None, write_type="append", service_account=None, confirm=False), update_scheduled_query(name, query=None, schedule=None, disabled=None), run_scheduled_query(name, run_time=None), delete_scheduled_query(name, confirm=False)

Contributing

Issues and pull requests welcome at the project repository.

See LICENSE file for details.