使用 Python 进行 GCP BigQuery 的 Google OAuth 2.0

Question

我正在寻找一个代码 sn-p，用于使用 python 连接到 GCP Big Query 服务来实现 oAuth 2.0 身份验证。

我正在使用 Google Cloud Shell 来编写 Python 代码。但是我收到了错误的访问令牌。

access_token = google.fetch_token(token_url=token_url,client_id=client_id,client_secret=client_secret,authorization_response=redirect_response).

我还需要自动化这个过程，所以需要避免手动粘贴redirect_response。

Answer 1

建议您使用 BigQuery Python 客户端库。 Pip 包google-cloud-bigquery 提供了这个。您还需要使用服务帐户 json 文件设置 GOOGLE_APPLICATION_CREDENTIALS。

使用此过程，您无需处理令牌生成和更新，因为此过程由后台的客户端库处理。

有关详细说明，请参阅BigQuery Client Libraries Python 部分。

Answer 2

在 BigQuery Client Libraries 中记录了如何从 GCP 控制台和 Command Line 设置身份验证。

要使用BigQuery API library，您需要验证您的服务帐户。 gcloud 命令gcloud iam service-accounts keys create [FILE_NAME].json --iam-account [NAME]@[PROJECT_ID].iam.gserviceaccount.com 会生成一个 JSON 密钥文件，其中包含必要的私有信息（如您的 project_id、私有密钥等）。

进行 BigQuery API 调用时，您需要向应用代码提供此类凭据。可以通过设置环境变量GOOGLE_APPLICATION_CREDENTIALS指向服务账号JSON文件的路径来实现

export GOOGLE_APPLICATION_CREDENTIALS="PATH/TO/SERVICE_ACCOUNT.json"

但是，这仅在您当前的 shell 会话期间有效，因此如果此会话过期或您打开一个新会话，您将需要再次设置此变量。另一种验证凭据的方法是在 Python 脚本中使用方法
google.oauth2.Credentials.from_service_account_file。

在以下 Python 代码中，服务帐户使用方法 google.oauth2.Credentials.from_service_account_file 进行身份验证，从位于 Google Cloud Storage 中的 CSV 文件生成一个新的 BigQuery 表，并将新数据插入到该表中。

from google.cloud import bigquery
from google.oauth2 import service_account

# Path to the service account credentials
key_path = "/PATH/TO/SERVICE-ACCOUNT.json"
credentials = service_account.Credentials.from_service_account_file(
    key_path,
    scopes=["https://www.googleapis.com/auth/cloud-platform"],
)

# Instantiation of the BigQuery client
bigquery_client = bigquery.Client()

GCS_URI    = "gs://MY_BUCKET/MY_CSV_FILE"
DATASET_ID = "MY_DATASET"
TABLE_ID   = "MY_TABLE"

def bq_insert_from_gcs(target_uri = GCS_URI, dataset_id = DATASET_ID, table_id = TABLE_ID):
    """This method inserts a CSV file stored in GCS into a BigQuery Table."""

    dataset_ref = bigquery_client.dataset(dataset_id)

    job_config = bigquery.LoadJobConfig()
    # Schema autodetection enabled
    job_config.autodetect = True
    # Skipping first row which correspnds to the field names
    job_config.skip_leading_rows = 1
    # Format of the data in GCS
    job_config.source_format = bigquery.SourceFormat.CSV
    load_job = bigquery_client.load_table_from_uri(target_uri,\
                                                   dataset_ref.table(table_id),\
                                                   job_config=job_config)\

    print('Starting job {}'.format(load_job.job_id))
    print('Loading file {} into the Bigquery table {}'.format(target_uri, table_id))

    load_job.result()
    return 'Job finished.\n'


def bq_insert_to_table(rows_to_insert, dataset_id = DATASET_ID, table_id= TABLE_ID):
    """This method inserts rows into a BigQuery table"""

    # Prepares a reference to the dataset and table
    dataset_ref = bigquery_client.dataset(dataset_id)
    table_ref = dataset_ref.table(table_id)
    # API request to get table call
    table = bigquery_client.get_table(table_ref)

    # API request to insert the rows_to_insert
    print("Inserting rows into BigQuery table {}".format(table_id))
    errors = bigquery_client.insert_rows(table, rows_to_insert)
    assert errors == []


bq_insert_from_gcs()

rows_to_insert = [( u'Alice', u'cat'),\
                  (u'John', u'dog')]
bq_insert_to_table(rows_to_insert)

另外，我强烈建议您使用 Python 3 来实现您的脚本，因为从 2020 年 1 月 1 日起，google-cloud-bigquery 将不再支持 Python 2。

Answer 3

您将需要导出到 json 的服务帐户的凭据。 GCP -> IAM 和管理员 -> 服务帐户，在三个小点下，您将找到您帐户的创建密钥。

如前面的答案所述，您还需要BigQuery library

那么这样的事情就可以了

from google.cloud import bigquery
from google.oauth2 import service_account

def BigQuery():
  try:
    credentials = service_account.Credentials.from_service_account_file(
      '/Credentials.json')
    project_id = '[project_id]
    client = bigquery.Client(credentials= credentials,project=project_id)

    query = ('SELECT Column1, Column2 FROM `{}.{}.{}` limit 20'.format('[project_id]','[dataset]','[table]'))
    query_job = client.query(query)
    results = query_job.result()
    for row in results:
      print('Column1 1 : {}, Column 2: {}'.format(row.Column1, row.Column2))
  except:
    print('Error!')



if __name__ == '__main__':
  BigQuery()