Cloud Storage 触发器

您可以触发一个函数来响应 Cloud Storage 中上传、更新或删除文件和文件夹的操作。

本页中的示例基于一个在图片文件上传到 Cloud Storage 时触发的示例函数。此示例函数演示了如何访问事件属性、如何将文件下载到 Cloud Functions 实例，以及其他有关处理 Cloud Storage 事件的基础操作。

导入所需的模块

首先，导入处理 Cloud Storage 事件所需的模块：

Node.js

 const {onObjectFinalized} = require("firebase-functions/v2/storage");index.js

Python

 from firebase_functions import storage_fnmain.py

如需构建完整示例，还要添加 Firebase Admin SDK 和图片处理工具的依赖项：

Node.js

 const {initializeApp} = require("firebase-admin/app");
const {getStorage} = require("firebase-admin/storage");
const logger = require("firebase-functions/logger");
const path = require("path");

// library for image resizing
const sharp = require("sharp");

initializeApp();index.js

Python

 import io
import pathlib

from PIL import Image

from firebase_admin import initialize_app

initialize_app()
from firebase_admin import storagemain.py

限定 Cloud Storage 函数的范围

您可以使用以下形式的脚本将您的函数范围限定在特定的 Cloud Storage 存储桶，并设置需要的所有选项：

Node.js

// scope handler to a specific bucket, using storage options parameter
export archivedopts = onObjectArchived({ bucket: "myBucket" }, (event) => {
  //…
});

Python

# Scope handler to a specific bucket using storage options parameter
@storage_fn.on_object_archived(bucket="myBucket")
def archived_bucket(event: storage_fn.CloudEvent[storage_fn.StorageObjectData]):
    # ...

与此相对，下面的缩略图生成器示例函数的范围则是限定在项目的默认存储桶：

Node.js

exports.generateThumbnail = onObjectFinalized({cpu: 2}, async (event) => {
// ...
});

Python

@storage_fn.on_object_archived()
def generatethumbnail(event: storage_fn.CloudEvent[storage_fn.StorageObjectData]):
    # ...

设置函数位置

位置之间的不匹配可能会导致部署失败。此外，Cloud Storage 存储桶的位置与函数位置之间的距离还可能会导致严重的网络延迟。为避免这些情况，请指定函数位置，使其通过以下某种方式匹配存储桶/触发器位置：

函数位置与触发器位置相同
函数位置位于触发器位置范围内（当触发器区域为双区域/多区域时）
如果触发器区域设置为 us-central1，函数可能位于任何位置

处理 Cloud Storage 事件

响应 Cloud Storage 事件可使用以下处理程序：

Node.js

onObjectArchived：仅在存储桶已启用对象版本控制时发送。此事件表明某个对象的当前版本已变为归档版本，原因可能是该对象已归档，或者已被新上传的同名对象覆盖。
onObjectDeleted：当某个对象被永久删除时发送。这包括系统根据存储桶的生命周期配置所覆盖或删除的对象。对于启用了对象版本控制的存储桶，如果某个对象已归档（请参阅 onArchive），则系统不会发送此事件，即使通过 storage.objects.delete 方法进行归档也是如此。
onObjectFinalized：当存储桶中成功创建了新对象（或现有对象的新实例）时发送。这包括复制或重写现有对象。上传失败不会触发此事件。
onMetadataUpdated：当现有对象的元数据发生更改时发送。

Python

on_object_archived：仅在存储桶已启用对象版本控制时发送。此事件表明某个对象的当前版本已变为归档版本，原因可能是该对象已归档，或者已被新上传的同名对象覆盖。
on_object_deleted：当某个对象被永久删除时发送。这包括系统根据存储桶的生命周期配置所覆盖或删除的对象。对于启用了对象版本控制的存储桶，如果某个对象已归档（请参阅 onArchive），则系统不会发送此事件，即使通过 storage.objects.delete 方法进行归档也是如此。
on_object_finalized：当存储桶中成功创建了新对象（或现有对象的新实例）时发送。这包括复制或重写现有对象。上传失败不会触发此事件。
on_metadata_updated：当现有对象的元数据发生更改时发送。

访问 Cloud Storage 对象属性

Cloud Functions 会显示一些 Cloud Storage 对象属性，例如所更新文件的对象大小及内容类型。每当对象的元数据发生变化时，metageneration 属性都会递增。对于新对象，metageneration 值为 1。

Node.js

const fileBucket = event.data.bucket; // Storage bucket containing the file.
const filePath = event.data.name; // File path in the bucket.
const contentType = event.data.contentType; // File content type.index.js

Python

bucket_name = event.data.bucket
file_path = pathlib.PurePath(event.data.name)
content_type = event.data.content_typemain.py

缩略图生成器示例使用了这样的一些属性来检测要让函数返回的退出情况：

Node.js

// Exit if this is triggered on a file that is not an image.
if (!contentType.startsWith("image/")) {
  return logger.log("This is not an image.");
}
// Exit if the image is already a thumbnail.
const fileName = path.basename(filePath);
if (fileName.startsWith("thumb_")) {
  return logger.log("Already a Thumbnail.");
}index.js

Python

# Exit if this is triggered on a file that is not an image.
if not content_type or not content_type.startswith("image/"):
    print(f"This is not an image. ({content_type})")
    return

# Exit if the image is already a thumbnail.
if file_path.name.startswith("thumb_"):
    print("Already a thumbnail.")
    returnmain.py

下载、转换和上传文件

在某些情况下，您可能不需要从 Cloud Storage 下载文件。但是，执行密集型任务（例如根据 Cloud Storage 中存储的文件生成缩略图）时，您需要将文件下载到函数实例（即运行代码的虚拟机）。

可以将 Cloud Functions 与一些图片处理程序（例如适用于 Node.js 的 sharp 以及适用于 Python 的 Pillow）结合使用，对图片文件执行各种操作。以下示例说明了如何为上传的图片文件创建缩略图：

Node.js

/**
 * When an image is uploaded in the Storage bucket,
 * generate a thumbnail automatically using sharp.
 */
exports.generateThumbnail = onObjectFinalized({cpu: 2}, async (event) => {

  const fileBucket = event.data.bucket; // Storage bucket containing the file.
  const filePath = event.data.name; // File path in the bucket.
  const contentType = event.data.contentType; // File content type.

  // Exit if this is triggered on a file that is not an image.
  if (!contentType.startsWith("image/")) {
    return logger.log("This is not an image.");
  }
  // Exit if the image is already a thumbnail.
  const fileName = path.basename(filePath);
  if (fileName.startsWith("thumb_")) {
    return logger.log("Already a Thumbnail.");
  }

  // Download file into memory from bucket.
  const bucket = getStorage().bucket(fileBucket);
  const downloadResponse = await bucket.file(filePath).download();
  const imageBuffer = downloadResponse[0];
  logger.log("Image downloaded!");

  // Generate a thumbnail using sharp.
  const thumbnailBuffer = await sharp(imageBuffer).resize({
    width: 200,
    height: 200,
    withoutEnlargement: true,
  }).toBuffer();
  logger.log("Thumbnail created");

  // Prefix 'thumb_' to file name.
  const thumbFileName = `thumb_${fileName}`;
  const thumbFilePath = path.join(path.dirname(filePath), thumbFileName);

  // Upload the thumbnail.
  const metadata = {contentType: contentType};
  await bucket.file(thumbFilePath).save(thumbnailBuffer, {
    metadata: metadata,
  });
  return logger.log("Thumbnail uploaded!");
});index.js

将文件下载到 Cloud Functions 实例的临时目录中。在此位置，您可以根据需要处理文件，然后将处理的结果上传到 Cloud Storage。执行异步任务时，请确保在回调中返回 JavaScript promise。

Python

@storage_fn.on_object_finalized()
def generatethumbnail(event: storage_fn.CloudEvent[storage_fn.StorageObjectData]):
    """When an image is uploaded in the Storage bucket, generate a thumbnail
    automatically using Pillow."""

    bucket_name = event.data.bucket
    file_path = pathlib.PurePath(event.data.name)
    content_type = event.data.content_type

    # Exit if this is triggered on a file that is not an image.
    if not content_type or not content_type.startswith("image/"):
        print(f"This is not an image. ({content_type})")
        return

    # Exit if the image is already a thumbnail.
    if file_path.name.startswith("thumb_"):
        print("Already a thumbnail.")
        return

    bucket = storage.bucket(bucket_name)

    image_blob = bucket.blob(str(file_path))
    image_bytes = image_blob.download_as_bytes()
    image = Image.open(io.BytesIO(image_bytes))

    image.thumbnail((200, 200))
    thumbnail_io = io.BytesIO()
    image.save(thumbnail_io, format="png")
    thumbnail_path = file_path.parent / pathlib.PurePath(f"thumb_{file_path.stem}.png")
    thumbnail_blob = bucket.blob(str(thumbnail_path))
    thumbnail_blob.upload_from_string(thumbnail_io.getvalue(), content_type="image/png")main.py

上面这段代码会为保存在临时目录中的图片创建大小为 200x200 的缩略图，然后将缩略图传回 Cloud Storage。