将 App Engine Blobstore 迁移到 Cloud Storage

本指南介绍了如何从 App Engine Blobstore 迁移到 Cloud Storage。

Cloud Storage 与 App Engine Blobstore 类似，您可以使用 Cloud Storage 提供大型数据对象 (blob)，例如视频或图片文件，并让您的用户上传大型数据文件。虽然 App Engine Blobstore 只能通过 App Engine 旧版捆绑服务进行访问，但 Cloud Storage 是一种独立 Google Cloud 产品，可通过 Cloud 客户端库访问它。Cloud Storage 可为您的应用提供更现代化的对象存储解决方案，并在之后让您可以灵活地迁移到 Cloud Run 或其他 Google Cloud 应用托管平台。

对于 2016 年 11 月之后创建的 Google Cloud 项目，Blobstore 在后台使用 Cloud Storage 存储桶。这意味着，将应用迁移到 Cloud Storage 后，这些现有 Cloud Storage 存储桶中的所有现有对象和权限都会保持不变。您还可以开始使用适用于 Cloud Storage 的 Cloud 客户端库访问这些现有存储桶。

主要区别和相似之处

Cloud Storage 不包括以下 Blobstore 依赖项和限制：

Python 2 版 Blobstore API 依赖于 webapp。
Python 3 版 Blobstore API 通过实用程序类来使用 Blobstore 处理程序。
对于 Blobstore，可以上传到 Blobstore 的文件数量的上限为 500。您可以在 Cloud Storage 存储桶中创建的对象数量不受限制。

Cloud Storage 不支持：

Blobstore 处理程序类
Blobstore 对象

Cloud Storage 与 App Engine Blobstore 的相似之处：

能够在运行时环境中读取和写入大型数据对象，以及存储和传送静态大型数据对象，例如影片、图片或其他静态内容。Cloud Storage 的对象大小限制为 5 TiB。
可让您将对象存储在 Cloud Storage 存储桶中。
具有免费层级。

准备工作

您应该查看并了解 Cloud Storage 价格和配额：
- Cloud Storage 是付费服务，根据数据的存储类别和存储桶的位置计算自己的数据存储价格。
- Cloud Storage 配额与 App Engine Blobstore 配额和限制存在一些差异，这可能会影响您的 App Engine 请求配额。
已有使用 Blobstore 的 Python 2 或 Python 3 App Engine 应用。
本指南中的示例展示了一个使用 Flask 框架迁移到 Cloud Storage 的应用。请注意，在迁移到 Cloud Storage 时，您可以使用任何 Web 框架，包括保留 webapp2。

概览

概括来讲，从 App Engine Blobstore 迁移到 Cloud Storage 的过程包括以下步骤：

更新配置文件
更新 Python 应用：
- 更新 Web 框架
- 导入并初始化 Cloud Storage
- 更新 Blobstore 处理程序
- 可选：如果使用 Cloud NDB 或 App Engine NDB，请更新数据模型
测试和部署应用

更新配置文件

在修改应用代码以从 Blobstore 迁移到 Cloud Storage 之前，请更新配置文件以使用 Cloud Storage 库。

更新 app.yaml 文件。按照适用于您的 Python 版本的说明操作：
Python 2
对于 Python 2 应用：
1. 移除 handlers 部分，以及 libraries 部分中的所有不必要的 Web 应用依赖项。
2. 如果您使用 Cloud 客户端库，请添加最新版本的 grpcio 和 setuptools 库。
3. 添加 ssl 库，因为 Cloud Storage 要求这样做。
以下是一个示例 app.yaml 文件，其中包含所做的更改：
```
runtime: python27
threadsafe: yes
api_version: 1

handlers:
- url: /.*
  script: main.app

libraries:
- name: grpcio
  version: latest
- name: setuptools
  version: latest
- name: ssl
  version: latest
```
Python 3
对于 Python 3 应用，请删除除 runtime 元素之外的所有行。例如：
```
runtime: python310 # or another support version
```
Python 3 运行时会自动安装库，因此您无需指定过往 Python 2 运行时中的内置库。如果 Python 3 应用在迁移到 Cloud Storage 时使用其他旧版捆绑服务，请保持 app.yaml 文件不变。
更新 requirements.txt 文件。按照适用于您的 Python 版本的说明操作：
Python 2
将适用于 Cloud Storage 的 Cloud 客户端库添加到 requirements.txt 文件中的依赖项列表。
```
google-cloud-storage
```
然后，运行 pip install -t lib -r requirements.txt 以更新应用的可用库列表。
Python 3
将适用于 Cloud Storage 的 Cloud 客户端库添加到 requirements.txt 文件中的依赖项列表。
```
google-cloud-storage
```
App Engine 会在 Python 3 运行时的应用部署期间自动安装这些依赖项，因此如果存在 lib 文件夹，请删除它。

对于 Python 2 应用，如果您的应用使用内置库或复制的库，则必须在 appengine_config.py 文件中指定以下路径：

import pkg_resources
from google.appengine.ext import vendor

# Set PATH to your libraries folder.
PATH = 'lib'
# Add libraries installed in the PATH folder.
vendor.add(PATH)
# Add libraries to pkg_resources working set to find the distribution.
pkg_resources.working_set.add_entry(PATH)

更新 Python 应用

修改配置文件后，请更新 Python 应用。

更新 Python 2 Web 框架

对于使用 webapp2 框架的 Python 2 应用，建议将已过时的 webapp2 框架迁走。请参阅运行时支持时间表以获取 Python 2 支持终止日期。

您可以迁移到其他网络框架，例如 Flask、Django 或 WSGI。由于 Cloud Storage 排除了 webapp2 的依赖项，并且 Blobstore 处理程序不受支持，因此您可以删除或替换其他与 webapp 相关的库。

如果您选择继续使用 webapp2，请注意本指南中的示例会将 Cloud Storage 与 Flask 搭配使用。

如果除了 Cloud Storage 之外，您还想要使用 Google Cloud 服务，或者为了获取最新运行时版本的访问权限，您应考虑将应用升级到 Python 3 运行时。如需了解详情，请参阅 Python 2 到 Python 3 迁移概览。

导入并初始化 Cloud Storage

通过更新导入和初始化行来修改应用文件：

移除 Blobstore import 语句，如下所示：

import webapp2
from google.appengine.ext import blobstore
from google.appengine.ext.webapp import blobstore_handlers

为 Cloud Storage 和 Google 身份验证库添加 import 语句，如下所示：
```
import io
from flask import (Flask, abort, redirect, render_template,
request, send_file, url_for)
from google.cloud import storage
import google.auth
```
需要使用 Google 身份验证库才能为 Cloud Storage 获取 Blobstore 中使用的项目 ID。导入 Cloud NBD 等其他库，前提是适用于您的应用。
为 Cloud Storage 创建新客户端并指定 Blobstore 中使用的存储桶。例如：
```
gcs_client = storage.Client()
_, PROJECT_ID = google.auth.default()
BUCKET = '%s.appspot.com' % PROJECT_ID
```
对于 2016 年 11 月之后的 Google Cloud 项目，Blobstore 会将数据写入以应用网址命名的 Cloud Storage 存储桶，并遵循 PROJECT_ID.appspot.com 格式。您可以使用 Google 身份验证获取项目 ID，以指定用于在 Blobstore 中存储 blob 的 Cloud Storage 存储桶。

更新 Blobstore 处理程序

由于 Cloud Storage 不支持 Blobstore 上传和下载处理程序，因此您需要结合使用 Cloud Storage 功能、io 标准库模块、Web 框架和 Python 实用程序来上传和下载 Cloud Storage 中的对象 (blob)。

以下示例演示了如何使用 Flask 作为示例 Web 框架来更新 Blobstore 处理程序：

将 Blobstore 上传处理程序类替换为 Flask 中的上传函数。按照适用于您的 Python 版本的说明操作：

Python 2

Python 2 中的 Blobstore 处理程序是 webapp2 类，如以下 Blobstore 示例所示：

class UploadHandler(blobstore_handlers.BlobstoreUploadHandler):
    'Upload blob (POST) handler'
    def post(self):
        uploads = self.get_uploads()
        blob_id = uploads[0].key() if uploads else None
        store_visit(self.request.remote_addr, self.request.user_agent, blob_id)
        self.redirect('/', code=307)
...
app = webapp2.WSGIApplication([
    ('/', MainHandler),
    ('/upload', UploadHandler),
    ('/view/([^/]+)?', ViewBlobHandler),
], debug=True)

如需使用 Cloud Storage，请执行以下操作：

将 Webapp 上传类替换为 Flask 上传函数。
将上传处理程序和路由替换为使用路由修饰的 Flask POST 方法。

更新后的代码示例：

@app.route('/upload', methods=['POST'])
def upload():
    'Upload blob (POST) handler'
    fname = None
    upload = request.files.get('file', None)
    if upload:
        fname = secure_filename(upload.filename)
        blob = gcs_client.bucket(BUCKET).blob(fname)
        blob.upload_from_file(upload, content_type=upload.content_type)
    store_visit(request.remote_addr, request.user_agent, fname)
    return redirect(url_for('root'), code=307)

在更新后的 Cloud Storage 代码示例中，应用现在通过其对象名称 (fname) 而不是 blob_id 来标识对象工件。路由也会在应用文件底部发生。

为了获取上传的对象，Blobstore 的 get_uploads() 方法已替换为 Flask 的 request.files.get() 方法。在 Flask 中，您可以使用 secure_filename() 方法获取文件的名称，而不使用路径字符（例如 /），并通过使用 gcs_client.bucket(BUCKET).blob(fname) 指定存储桶名称和对象名称来标识对象。

Cloud Storage upload_from_file() 调用执行上传，如更新后的示例所示。

Python 3

Python 3 版 Blobstore 中的上传处理程序类是一个实用程序类，需要使用 WSGI environ 字典作为输入参数，如以下 Blobstore 示例所示：

class UploadHandler(blobstore.BlobstoreUploadHandler):
    'Upload blob (POST) handler'
    def post(self):
        uploads = self.get_uploads(request.environ)
        if uploads:
            blob_id = uploads[0].key()
            store_visit(request.remote_addr, request.user_agent, blob_id)
        return redirect('/', code=307)
...
@app.route('/upload', methods=['POST'])
def upload():
    """Upload handler called by blobstore when a blob is uploaded in the test."""
    return UploadHandler().post()

如需使用 Cloud Storage，请将 Blobstore 的 get_uploads(request.environ) 方法替换为 Flask 的 request.files.get() 方法。

更新后的代码示例：

@app.route('/upload', methods=['POST'])
def upload():
    'Upload blob (POST) handler'
    fname = None
    upload = request.files.get('file', None)
    if upload:
        fname = secure_filename(upload.filename)
        blob = gcs_client.bucket(BUCKET).blob(fname)
        blob.upload_from_file(upload, content_type=upload.content_type)
    store_visit(request.remote_addr, request.user_agent, fname)
    return redirect(url_for('root'), code=307)

在更新后的 Cloud Storage 代码示例中，应用现在通过其对象名称 (fname) 而不是 blob_id 来标识对象工件。路由也会在应用文件底部发生。

Cloud Storage upload_from_file() 方法执行上传，如更新后的示例所示。

将 Blobstore 下载处理程序类替换为 Flask 中的下载函数。按照适用于您的 Python 版本的说明操作：

Python 2

以下下载处理程序示例展示了使用 webapp2 的 BlobstoreDownloadHandler 类的用法：

class ViewBlobHandler(blobstore_handlers.BlobstoreDownloadHandler):
    'view uploaded blob (GET) handler'
    def get(self, blob_key):
        self.send_blob(blob_key) if blobstore.get(blob_key) else self.error(404)
...
app = webapp2.WSGIApplication([
    ('/', MainHandler),
    ('/upload', UploadHandler),
    ('/view/([^/]+)?', ViewBlobHandler),
], debug=True)

如需使用 Cloud Storage，请执行以下操作：

更新 Blobstore 的 send_blob() 方法以使用 Cloud Storage 的 download_as_bytes() 方法。
将路由从 webapp2 更改为 Flask。

更新后的代码示例：

@app.route('/view/<path:fname>')
def view(fname):
    'view uploaded blob (GET) handler'
    blob = gcs_client.bucket(BUCKET).blob(fname)
    try:
        media = blob.download_as_bytes()
    except exceptions.NotFound:
        abort(404)
    return send_file(io.BytesIO(media), mimetype=blob.content_type)

在更新后的 Cloud Storage 代码示例中，Flask 会修饰 Flask 函数中的路由并使用 '/view/<path:fname>' 来标识对象。Cloud Storage 通过对象名称和存储桶名称来标识 blob 对象，并使用 download_as_bytes() 方法以字节形式下载对象，而不是使用 Blobstore 中的 send_blob 方法。如果未找到工件，则应用会返回 HTTP 404 错误。

Python 3

与上传处理程序一样，Python 3 版 Blobstore 中的下载处理程序类是一个实用程序类，需要使用 WSGI environ 字典作为输入参数，如以下 Blobstore 示例所示：

class ViewBlobHandler(blobstore.BlobstoreDownloadHandler):
    'view uploaded blob (GET) handler'
    def get(self, blob_key):
        if not blobstore.get(blob_key):
            return "Photo key not found", 404
        else:
            headers = self.send_blob(request.environ, blob_key)

        # Prevent Flask from setting a default content-type.
        # GAE sets it to a guessed type if the header is not set.
        headers['Content-Type'] = None
        return '', headers
...
@app.route('/view/<blob_key>')
def view_photo(blob_key):
    """View photo given a key."""
    return ViewBlobHandler().get(blob_key)

如需使用 Cloud Storage，请将 Blobstore 的 send_blob(request.environ, blob_key) 替换为 Cloud Storage 的 blob.download_as_bytes() 方法。

更新后的代码示例：

@app.route('/view/<path:fname>')
def view(fname):
    'view uploaded blob (GET) handler'
    blob = gcs_client.bucket(BUCKET).blob(fname)
    try:
        media = blob.download_as_bytes()
    except exceptions.NotFound:
        abort(404)
    return send_file(io.BytesIO(media), mimetype=blob.content_type)

在更新后的 Cloud Storage 代码示例中，blob_key 已替换为 fname，而 Flask 使用 '/view/<path:fname>' 网址来标识对象。gcs_client.bucket(BUCKET).blob(fname) 方法用于查找文件名和存储桶名称。 Cloud Storage 的 download_as_bytes() 方法会以字节形式下载对象，而不是使用 Blobstore 中的 send_blob() 方法。

如果应用使用主处理程序，请将 MainHandler 类替换为 Flask 中的 root() 函数。按照适用于您的 Python 版本的说明操作：

Python 2

下面是一个使用 Blobstore 的 MainHandler 类的示例：

class MainHandler(BaseHandler):
    'main application (GET/POST) handler'
    def get(self):
        self.render_response('index.html',
                upload_url=blobstore.create_upload_url('/upload'))

    def post(self):
        visits = fetch_visits(10)
        self.render_response('index.html', visits=visits)

app = webapp2.WSGIApplication([
    ('/', MainHandler),
    ('/upload', UploadHandler),
    ('/view/([^/]+)?', ViewBlobHandler),
], debug=True)

如需使用 Cloud Storage，请执行以下操作：

移除 MainHandler(BaseHandler) 类，因为 Flask 会为您处理路由。
使用 Flask 简化 Blobstore 代码。
移除末尾的 webapp 路由。

更新后的代码示例：

@app.route('/', methods=['GET', 'POST'])
def root():
    'main application (GET/POST) handler'
    context = {}
    if request.method == 'GET':
        context['upload_url'] = url_for('upload')
    else:
        context['visits'] = fetch_visits(10)
    return render_template('index.html', **context)

Python 3

如果您使用的是 Flask，则您没有 MainHandler 类，但如果使用 blobstore，则需要更新 Flask 根函数。以下示例使用 blobstore.create_upload_url('/upload') 函数：

@app.route('/', methods=['GET', 'POST'])
def root():
    'main application (GET/POST) handler'
    context = {}
    if request.method == 'GET':
        context['upload_url'] = blobstore.create_upload_url('/upload')
    else:
        context['visits'] = fetch_visits(10)
    return render_template('index.html', **context)

如需使用 Cloud Storage，请将 blobstore.create_upload_url('/upload') 函数替换为 Flask 的 url_for() 方法，以获取 upload() 函数的网址。

更新后的代码示例：

@app.route('/', methods=['GET', 'POST'])
def root():
    'main application (GET/POST) handler'
    context = {}
    if request.method == 'GET':
        context['upload_url'] = url_for('upload') # Updated to use url_for
    else:
        context['visits'] = fetch_visits(10)
    return render_template('index.html', **context)

测试和部署应用

通过本地开发服务器，您可以测试应用是否运行，但在部署新版本之前，您无法测试 Cloud Storage，因为所有 Cloud Storage 请求都需要通过互联网发送到实际的 Cloud Storage 存储桶。请参阅测试和部署应用，以了解如何在本地运行应用。然后，部署新版本，确认应用的外观与之前相同。

使用 App Engine NDB 或 Cloud NDB 的应用

如果您的应用使用 App Engine NDB 或 Cloud NDB 来添加与 Blobstore 相关的属性，则必须更新 Datastore 数据模型。

更新数据模型

由于 Cloud Storage 不支持 NDB 中的 BlobKey 属性，因此您需要修改与 Blobstore 相关的行，才能使用 NDB、Web 框架或其他环境中的内置等效项。

如需更新数据模型，请执行以下操作：

在数据模型中找到使用 BlobKey 的行，如以下所示：

class Visit(ndb.Model):
    'Visit entity registers visitor IP address & timestamp'
    visitor   = ndb.StringProperty()
    timestamp = ndb.DateTimeProperty(auto_now_add=True)
    file_blob = ndb.BlobKeyProperty()

将 ndb.BlobKeyProperty() 替换为 ndb.StringProperty()：

class Visit(ndb.Model):
    'Visit entity registers visitor IP address & timestamp'
    visitor   = ndb.StringProperty()
    timestamp = ndb.DateTimeProperty(auto_now_add=True)
    file_blob = ndb.StringProperty() # Modified from ndb.BlobKeyProperty()

如果您还在迁移过程中从 App Engine NDB 升级到 Cloud NDB，请参阅 Cloud NDB 迁移指南，了解如何重构 NDB 代码以使用 Python 上下文管理器。

Datastore 数据模型的向后兼容性

在前面部分，将 ndb.BlobKeyProperty 替换为 ndb.StringProperty 会导致应用向后不兼容，这意味着应用将无法处理 Blobstore 创建的旧条目。如果您需要保留旧数据，请为新的 Cloud Storage 条目创建另一个字段，而不是更新 ndb.BlobKeyProperty 字段，并创建一个函数来对数据进行标准化。

在前面部分的示例中，请进行以下更改：

定义数据模型时，创建两个单独的属性字段。使用 file_blob 属性可标识 Blobstore 创建的对象，使用 file_gcs 属性可标识 Cloud Storage 创建的对象：

class Visit(ndb.Model):
    'Visit entity registers visitor IP address & timestamp'
    visitor   = ndb.StringProperty()
    timestamp = ndb.DateTimeProperty(auto_now_add=True)
    file_blob = ndb.BlobKeyProperty()  # backwards-compatibility
    file_gcs  = ndb.StringProperty()

找到引用新访问的行，如下所示：

def store_visit(remote_addr, user_agent, upload_key):
    'create new Visit entity in Datastore'
    with ds_client.context():
        Visit(visitor='{}: {}'.format(remote_addr, user_agent),
                file_blob=upload_key).put()

更改代码，以使 file_gcs 用于近期的条目。例如：

def store_visit(remote_addr, user_agent, upload_key):
    'create new Visit entity in Datastore'
    with ds_client.context():
        Visit(visitor='{}: {}'.format(remote_addr, user_agent),
                file_gcs=upload_key).put() # change file_blob to file_gcs for new requests

创建一个新函数来对数据进行标准化。以下示例展示如何使用提取、转换和加载 (ETL) 循环遍历所有访问，并采用访问者和时间戳数据来检查是否存在 file_gcs 或 file_gcs：

def etl_visits(visits):
    return [{
            'visitor': v.visitor,
            'timestamp': v.timestamp,
            'file_blob': v.file_gcs if hasattr(v, 'file_gcs') \
                    and v.file_gcs else v.file_blob
            } for v in visits]

找到引用 fetch_visits() 函数的行：

@app.route('/', methods=['GET', 'POST'])
def root():
    'main application (GET/POST) handler'
    context = {}
    if request.method == 'GET':
        context['upload_url'] = url_for('upload')
    else:
        context['visits'] = fetch_visits(10)
    return render_template('index.html', **context)

将 fetch_visits() 封装在 etl_visits() 函数内，例如：

@app.route('/', methods=['GET', 'POST'])
def root():
    'main application (GET/POST) handler'
    context = {}
    if request.method == 'GET':
        context['upload_url'] = url_for('upload')
    else:
        context['visits'] = etl_visits(fetch_visits(10)) # etl_visits wraps around fetch_visits
    return render_template('index.html', **context)

示例

如需查看如何将 Python 2 应用迁移到 Cloud Storage 的示例，请比较 GitHub 中的 Python 2 版 Blobstore 代码示例和 Cloud Storage 代码示例。
如需查看如何将 Python 3 应用迁移到 Cloud Storage 的示例，请比较 GitHub 中的 Python 3 版 Blobstore 代码示例和 Cloud Storage 代码示例。

后续步骤

如需查看实操教程，请参阅从 App Engine Blobstore 迁移到 Python 版 Cloud Storage Codelab。
了解如何在 Cloud Storage 中存储和传送静态文件。
如需了解详情，请参阅 Cloud Storage 文档。