第 3 章 过滤的集成资源

# os.getenv(var) used to fetch secrets shared with the function
CLIENT_ID = os.getenv('client_id')
CLIENT_SECRET = os.getenv('client_secret')

token_url = 'https://sso.redhat.com/auth/realms/redhat-external/protocol/openid-connect/token'

token_headers = {'Content-Type': 'application/x-www-form-urlencoded'}
token_data = {"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "grant_type": "client_credentials"}
access_token = requests.post(token_url, headers=token_headers, data=token_data).json().get("access_token")

storage_client = storage.Client()
bucket = storage_client.bucket(“my_bucket”)
for rows in query_job:
    csv_file = "my_report_location/2025-04-01.csv"
    blob = bucket.blob(csv_file)
    with blob.open(mode='w') as f:
        writer = csv.writer(f)
        writer.writerow([my_col_1, my_col_col2])
        writer.writerows(rows)

def batch(iterable, n):
"""Yields successive n-sized chunks from iterable"""
	it = iter(iterable)
	while chunk := tuple(islice(it, n)):
    	yield chunk

for i, rows in enumerate(batch(query_job, BATCH_SIZE)):