第 3 章 过滤的集成资源

import csv
from google.cloud import storage

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, 200000)):
    # write csv file code block #

import os
import requests

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

# Get access token
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")

# Send reports to Red Hat
json_data = {"source": 0, "reports_list": ["my-file1.csv", "my-file-2.csv"], "bill_year": "2025", "bill_month": "07"}
headers = {'Authorization': f'Bearer {access_token}', 'Accept': 'application/json'}
requests.post("https://console.redhat.com/api/cost-management/v1/ingress/reports/", json=json_data, headers=headers)