Read input data with a BigQuery client

Create the client

Use the official Google Cloud client and Application Default Credentials. Set the project explicitly so queries, billing, and audit logs are attributed to the intended environment.

from google.cloud import bigquery

client = bigquery.Client(project="my-gcp-project")

Run a parameterized query

Keep SQL fixed and pass values as query parameters. Select only the columns and rows needed to construct the input envelope.

from google.cloud import bigquery

sql = """
SELECT event_id, event_type, occurred_at, workflow_id, payload
FROM `my-gcp-project.events.incoming`
WHERE occurred_at >= @since
ORDER BY occurred_at
LIMIT 1000
"""

job_config = bigquery.QueryJobConfig(
    query_parameters=[
        bigquery.ScalarQueryParameter("since", "TIMESTAMP", since),
    ]
)

for row in client.query(sql, job_config=job_config).result():
    submit_input(normalize_bigquery_row(dict(row)))

Keep the adapter bounded

• Use a read-only service account with access limited to the required datasets.
• Set query timeouts and maximum bytes billed to bound cost and latency.
• Checkpoint a stable cursor such as occurred_at plus event_id so retries do not skip or duplicate rows.
• Validate every row against the input-envelope schema before submitting it to the runtime.
• Treat warehouse rows as evidence; only the validated runtime may advance workflow state.