python - 从 Cloud Function 将 GCS CSV 导入 Cloud SQL

Question

我想使用 Google Cloud Function 将数据从 GCS 中的 CSV 文件快速传输到 Cloud SQL 中的 Postgres 表。

理想情况下我会使用 GCP SDK 来执行此操作，但官方 documentation建议执行导入的唯一方法是 1)控制台、2)gcloud 命令或 3)curl。我在 Python 中使用 requests 采用了curl 方法。我在下面编写的代码(我省略了 Cloud Function 请求包装器)可以在我的笔记本电脑上运行，但不能在 Cloud Function 中运行。云函数完成时没有确认错误，但我的数据从未加载到表中，这与我从笔记本电脑运行代码时不同。

import google.auth.transport.requests
import json
import requests


credentials, project = google.auth.default()

"""
According to the docs, hitting the REST endpoint requires an 
access token to be passed in the request header. It was not
clear how to obtain an access token without resorting to
gcloud commands, but I finally stumbled across a solution
here: https://stackoverflow.com/a/55804230/554481

At this point in the code credentials.valid is False, and 
credentials.token is None, so I need to perform a refresh
to populate them.
"""
auth_req = google.auth.transport.requests.Request()
credentials.refresh(auth_req)

# Now you can use credentials.token
access_token = credentials.token

# Removed business details for SO post
instance_name = '<removed>'
bucket_name = '<removed>'
project_id = '<removed>'
gcs_path = '<removed>'
database = '<removed>'
table_name = '<removed>''

headers = {
    'Content-Type':'application/json',
    'Authorization':'Bearer '+str(access_token)
}
endpoint = 'https://www.googleapis.com/sql/v1beta4/projects/{project_id}/instances/{instance_name}/import'.format(
    project_id=project_id,
    instance_name=instance_name
)
gcs_location = 'gs://{bucket_name}/{gcs_path}'.format(
    bucket_name=bucket_name,
    gcs_path=gcs_path
)
json_payload = {
    "importContext": {
        "fileType": "CSV",
        "uri": gcs_location,
        "database": database,
        "csvImportOptions":{
            "table":table_name
        }
    }
}
requests.post(
    url=endpoint,
    data=json.dumps(json_payload),
    headers=headers
)

Answer 1

由于您没有检查响应，post 请求可能已成功执行，但仍返回错误代码。如果您看一下 Instance: import文档中，有一个使用发现客户端发出请求的示例:

from googleapiclient import discovery
from oauth2client.client import GoogleCredentials

credentials = GoogleCredentials.get_application_default()

service = discovery.build('sqladmin', 'v1beta4', credentials=credentials)

# Project ID of the project that contains the instance.
project = 'my-project'  # TODO: Update placeholder value.

# Cloud SQL instance ID. This does not include the project ID.
instance = 'my-instance'  # TODO: Update placeholder value.

instances_import_request_body = {
    # TODO: Add desired entries to the request body.
}

request = service.instances().import_(project=project, instance=instance, body=instances_import_request_body)
response = request.execute()

# TODO: Change code below to process the `response` dict:
pprint(response)

响应结果将显示操作是否成功。