在 python3 中自动执行 Cloud Firestore 导出

Question

我想使用基于flask的微服务设置我的Cloud Firestore数据库的自动备份服务，我需要使用的命令：

gcloud beta firestore export gs://[BUCKET_NAME]

这就是我想通过我的 App Engine 微服务运行的命令

@app.route('/backup', methods=["GET", "POST"])
def backup():

    subprocess.call('gcloud beta firestore export gs://bucket-name --async', shell=True)

    return f"Backup process started successfully, you can close this window. {datetime.now(timezone.utc)}"

但看起来并没有发生任何事情，我假设那是因为我的 App Engine 实例没有 CloudSDK。

这是我可以在 Cloud Function 中做的事情吗？

Answer 1

您无法在沙盒环境（appengine、函数）中执行系统调用。而且你不知道平台上安装了什么，这很危险/不一致。

您可以尝试使用 cloud run 或 app engine flex。但这不是一个真正的最佳实践。最好的方法是使用 Python 库以编程方式执行相同的操作。在任何情况下，底层结果都是一样的：API 调用。

Answer 2

这是一个示例应用，您可以使用 Google App Engine Cron 服务调用它。它基于node.js example in the docs:

app.yaml

runtime: python37

handlers:
- url: /.*
  script: auto

如果您已经部署了默认服务，请添加 target: cloud-firestore-admin 以创建新服务。

requirements.txt

Flask
google-api-python-client

google-api-python-client 简化了对 Cloud Firestore REST API 的访问。

ma​​in.py

import datetime
import os
from googleapiclient.discovery import build

from flask import Flask, request

app = Flask(__name__) 

@app.route('/cloud-firestore-export')
def export():
    # Deny if not from the GAE Cron Service
    assert request.headers['X-Appengine-Cron']
    # Deny if outputUriPrefix not set correctly
    outputUriPrefix = request.args.get('outputUriPrefix')
    assert outputUriPrefix and outputUriPrefix.startswith('gs://')
    # Use a timestamp in export file name
    timestamp = datetime.datetime.now().strftime('%Y%m%d-%H%M%S')
    if '/' not in outputUriPrefix[5:]:
      # Add a trailing slash if missing
      outputUriPrefix += '/' + timestamp
    else:
      outputUriPrefix += timestamp
    if 'collections' in request.args:
      collections = request.args.get('collections').split(",")
    else:
      collections = None

    body = {
        'collectionIds': collections,
        'outputUriPrefix': outputUriPrefix,
    }
    # Build REST API request for 
    # https://cloud.google.com/firestore/docs/reference/rest/v1/projects.databases/exportDocuments
    project_id = os.environ.get('GOOGLE_CLOUD_PROJECT')
    database_name = 'projects/{}/databases/(default)'.format(project_id)
    service = build('firestore', 'v1')
    service.projects().databases().exportDocuments(name=database_name, body=body).execute()
    return 'Operation started' 


if __name__ == '__main__':
    # This is used when running locally only. When deploying to Google App
    # Engine, a webserver process such as Gunicorn will serve the app. This
    # can be configured by adding an `entrypoint` to app.yaml.
    # Flask's development server will automatically serve static files in
    # the "static" directory. See:
    # http://flask.pocoo.org/docs/1.0/quickstart/#static-files. Once deployed,
    # App Engine itself will serve those files as configured in app.yaml.
    app.run(host='127.0.0.1', port=8080, debug=True)

cron.yaml

cron:
- description: "Daily Cloud Firestore Export"
  url: /cloud-firestore-export?outputUriPrefix=gs://BUCKET_NAME&collections=COLLECTIONS_LIST
  schedule: every 24 hours

如果您在 app.yaml 中部署到非默认服务，也请在此处添加：target: cloud-firestore-admin。

App Engine 服务帐户的访问权限

部署后，该应用会使用 GAE 服务帐号来授权导出请求。确保您的 GAE 服务帐号对 Cloud Firestore 和您的存储分区具有权限，请参阅：

https://cloud.google.com/firestore/docs/solutions/schedule-export#configure_access_permissions

Answer 3

一个最新的和更方便的方法可能如下所示

PS1。我刚刚运行了代码并按原样复制了

PS2。默认方法输入可能不会令人困惑。例如“{your-project-prefix}-develop”，可以是 gcp-project-id-develop 或 gcp-project-id-staging，您将在其中运行代码。

# pylint: disable=missing-module-docstring,missing-function-docstring,import-outside-toplevel
# pylint: disable=too-many-arguments,unused-argument,no-value-for-parameter
import os
import logging
from typing import List

from google.cloud import storage
from google.cloud.firestore_admin_v1.services.firestore_admin import client as admin_client
from google.cloud.firestore_admin_v1.types import firestore_admin

logger = logging.getLogger(__name__)


def create_storage_bucket_if_not_exists(
    gcp_project_id: str = '{your-project-prefix}-develop',
    bucket_name: str = '{your-project-prefix}-backup-firestore-develop'
):
    storage_client = storage.Client(project=gcp_project_id)
    bucket = storage_client.bucket(bucket_name)
    if not bucket.exists():
        bucket.storage_class = 'STANDARD'
        storage_client.create_bucket(bucket, location='us')


def get_client(
    service_account: str = '{your-service-account-path}'
):
    os.environ.unsetenv('GOOGLE_APPLICATION_CREDENTIALS')
    os.environ.pop('GOOGLE_APPLICATION_CREDENTIALS', None)
    os.environ.setdefault('GOOGLE_APPLICATION_CREDENTIALS', service_account)

    firestore_admin_client = admin_client.FirestoreAdminClient()
    return firestore_admin_client


def get_database_name(
    client: admin_client.FirestoreAdminClient,
    gcp_project_id: str = '{your-project-prefix}-develop'
):
    return client.database_path(gcp_project_id, '(default)')


def export_documents(
    client: admin_client.FirestoreAdminClient,
    database_name: str,
    collections: List[str] = None,
    bucket_name: str = '{your-project-prefix}-backup-firestore-develop',
    gcp_project_id: str = '{your-project-prefix}-develop'
):
    if collections is None:
        collections = []

    bucket = f'gs://{bucket_name}'
    request = firestore_admin.ExportDocumentsRequest(
        name=database_name,
        collection_ids=collections,
        output_uri_prefix=bucket
    )

    # it is gonna be finalized in the background - async
    operation = client.export_documents(
        request=request
    )
    return operation


def backup():
    client = get_client()
    database_name = get_database_name(client)
    create_storage_bucket_if_not_exists()
    export_documents(client, database_name, [])
    logger.info('Backup operation has been started!')


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    backup()

在这里您可以看到 GCS 存储桶下的备份目录