如何通过Python版Firebase Admin SDK加速Firebase Storage图片上传？

我编写了一段Python代码用于将图片上传至Firebase Storage：

import firebase_admin
from firebase_admin import credentials
from firebase_admin import storage
import os

cred = credentials.Certificate("certificate.json")
firebase_admin.initialize_app(cred, {
    'storageBucket' : 'AppName.appspot.com'
})
bucket = storage.bucket()

directory = '/Users/name/Desktop/TestOrdner/'
for foldername in os.listdir(directory):
    if str(foldername[0]) != '.':
        for image in os.listdir(directory + foldername):
            blob = bucket.blob(f'{foldername}/{image}')
            imagePath = directory + foldername + '/' + image
            blob.upload_from_filename(imagePath)

当前我有一个主文件夹，内部包含5个子文件夹，每个子文件夹中有4张图片，上传这些图片耗时15秒。若要上传150张图片，耗时会极长，请问有什么方法可以加快上传速度？

针对你的批量上传慢的问题，我有几个实用的优化方案，按效果优先级排序如下：

1. 采用并行上传（最直接的提速手段） #

你的当前代码是串行执行的，每次只能上传一张图片，完全没利用好网络带宽的潜力。用Python的concurrent.futures.ThreadPoolExecutor实现多线程并行上传，能让多张图片同时上传，大幅缩短总耗时。

修改后的代码示例：

import firebase_admin
from firebase_admin import credentials
from firebase_admin import storage
import os
from concurrent.futures import ThreadPoolExecutor

# 只初始化一次Firebase App，避免重复开销
cred = credentials.Certificate("certificate.json")
firebase_admin.initialize_app(cred, {
    'storageBucket' : 'AppName.appspot.com'
})
bucket = storage.bucket()

directory = '/Users/name/Desktop/TestOrdner/'

def upload_single_image(image_path, blob_path):
    """单个图片上传的封装函数"""
    blob = bucket.blob(blob_path)
    blob.upload_from_filename(image_path)

# 先收集所有需要上传的任务（路径映射）
upload_jobs = []
for foldername in os.listdir(directory):
    if foldername.startswith('.'):  # 跳过隐藏文件夹
        continue
    full_folder_path = os.path.join(directory, foldername)
    for image_filename in os.listdir(full_folder_path):
        full_image_path = os.path.join(full_folder_path, image_filename)
        target_blob_path = f'{foldername}/{image_filename}'
        upload_jobs.append( (full_image_path, target_blob_path) )

# 用线程池并行执行上传，线程数建议根据网络情况调整（10-20之间测试）
with ThreadPoolExecutor(max_workers=15) as executor:
    executor.map(lambda job: upload_single_image(*job), upload_jobs)

⚠️ 注意：线程数别设置得太夸张，否则可能触发Firebase的请求频率限制，建议先从10开始测试，再逐步调整到合适的数值。

2. 开启分片上传（针对大尺寸图片） #

如果你的单张图片体积较大（比如5MB以上），可以手动配置分片上传，将文件拆成多个小块并行上传。Firebase SDK的upload_from_filename默认已经支持分片，但你可以指定分片大小来优化：

# 在上传时添加chunk_size参数，比如设置为1MB（1024*1024字节）
blob.upload_from_filename(image_path, chunk_size=1024*1024)

这个优化对小文件提升不明显，但大文件能显著降低单文件的上传耗时。

3. 预压缩图片（从根源减少上传数据量） #

如果上传的图片不需要保留原始分辨率，可以先对图片进行压缩，减小文件体积，从根本上缩短上传时间。比如用PIL库做压缩：

from PIL import Image
import io

def compress_image(image_path, quality=80):
    """压缩图片，返回字节流"""
    with Image.open(image_path) as img:
        # 处理透明通道（如果是PNG的话）
        if img.mode in ("RGBA", "P"):
            img = img.convert("RGB")
        buffer = io.BytesIO()
        img.save(buffer, format="JPEG", quality=quality)
        buffer.seek(0)
        return buffer

# 修改上传函数为压缩后上传
def upload_single_image(image_path, blob_path):
    blob = bucket.blob(blob_path)
    compressed_img_buffer = compress_image(image_path)
    blob.upload_from_file(compressed_img_buffer, content_type='image/jpeg')

💡 提示：quality参数可以根据需求调整，数值越低压缩比越高，但画质损失也越大，建议在画质可接受的范围内尽量降低。

4. 优化连接复用（细节处的小提升） #

确保Firebase App只初始化一次（你的代码已经做到了），避免重复初始化带来的开销。另外，你可以自定义Storage Client的连接池配置，提升连接复用率：

from google.cloud.storage import Client

# 初始化时自定义Client，配置更大的连接池
client = Client(credentials=cred)
bucket = storage.bucket('AppName.appspot.com', client=client)

这个优化的提升幅度不算大，但在大规模上传时能减少重复建立连接的开销。

内容的提问来源于stack exchange，提问作者adri567