雲存儲觸發器


您可以觸發函數來回應 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_fn
main.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 storage
main.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 事件的處理程序可用：

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 公開許多 Clo​​s 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_type
main.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.")
    return
main.py

下載、轉換和上傳文件

在某些情況下，可能不需要從雲端儲存下載檔案。但是，要執行密集型任務（例如從 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 實例上的暫存目錄。在此位置，您可以根據需要處理文件，然後上傳到雲端儲存。執行非同步任務時，請確保在回呼中傳回 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。