如何通过Python实现Firebase新用户注册后自动发送SendGrid欢迎邮件？

阿华AIGC实验室

2026-5-29

嘿，我刚好有过类似的需求，给你一个纯Python的独立解决方案，完全不需要Cloud Functions或者Zapier，两种实现方式任你选：

核心思路

我们的目标是捕获新用户注册事件，然后自动调用SendGrid API发送欢迎邮件。这里提供两种方案：实时监听（用户注册后立即触发）和定时轮询（适合对实时性要求不高的场景），都用纯Python实现。

前置准备

首先安装所需依赖：

pip install firebase-admin sendgrid python-dotenv

然后创建一个.env文件管理敏感信息（避免硬编码密钥）：

FIREBASE_SERVICE_ACCOUNT_PATH=path/to/your/firebase-service-account.json
SENDGRID_API_KEY=your_sendgrid_api_key_here
SENDGRID_FROM_EMAIL=your-verified-sender-email@example.com

注意：Firebase服务账号密钥可以在Firebase控制台的「项目设置」→「服务账号」页面下载；SendGrid的发件邮箱必须经过验证，否则邮件会被归类为垃圾邮件。

方案1：实时监听Firebase数据库新增用户（推荐，实时触发）

如果你的应用在用户注册时会把用户数据写入Firestore或Realtime Database（大部分应用都会这么做），我们可以直接监听数据库的新增文档事件，一有新用户就发邮件。

代码示例（Firestore版本）

import os
import time
from dotenv import load_dotenv
import firebase_admin
from firebase_admin import credentials, firestore
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail

# 加载环境变量
load_dotenv()

# 初始化Firebase Admin SDK
cred = credentials.Certificate(os.getenv("FIREBASE_SERVICE_ACCOUNT_PATH"))
firebase_admin.initialize_app(cred)
db = firestore.client()

# 初始化SendGrid客户端
sg = SendGridAPIClient(os.getenv("SENDGRID_API_KEY"))
FROM_EMAIL = os.getenv("SENDGRID_FROM_EMAIL")

def send_welcome_email(to_email):
    """发送欢迎邮件的核心函数"""
    message = Mail(
        from_email=FROM_EMAIL,
        to_emails=to_email,
        subject="🎉 欢迎加入我们的平台！",
        html_content="<strong>很高兴你能来！</strong><p>感谢你注册我们的应用，期待你的精彩使用体验。</p>"
    )
    try:
        response = sg.send(message)
        print(f"✅ 欢迎邮件已发送到 {to_email}，状态码: {response.status_code}")
    except Exception as e:
        print(f"❌ 发送邮件失败 ({to_email}): {str(e)}")

def listen_for_new_users():
    """实时监听Firestore users集合的新增文档"""
    def on_snapshot(doc_snapshot, changes, read_time):
        for change in changes:
            if change.type.name == "ADDED":
                user_data = change.document.to_dict()
                user_email = user_data.get("email")
                if user_email:
                    send_welcome_email(user_email)
                else:
                    print(f"⚠️ 新用户 {change.document.id} 未提供邮箱信息")

    # 启动监听（监听users集合的所有变化）
    col_query = db.collection("users")
    watch = col_query.on_snapshot(on_snapshot)

    print("🚀 新用户监听服务已启动，按Ctrl+C停止...")
    try:
        while True:
            time.sleep(3600)  # 保持进程持续运行
    except KeyboardInterrupt:
        watch.unsubscribe()
        print("🛑 监听服务已停止")

if __name__ == "__main__":
    listen_for_new_users()

方案2：定时轮询Firebase Auth获取新用户（实现简单）

如果你的应用没有把用户数据写入数据库，或者不想用实时监听，可以定时查询Firebase Auth的用户列表，对比上次查询的时间，找出新注册的用户。

代码示例

import os
import time
from datetime import datetime, timezone
from dotenv import load_dotenv
import firebase_admin
from firebase_admin import credentials, auth
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail

# 加载环境变量和初始化服务
load_dotenv()
cred = credentials.Certificate(os.getenv("FIREBASE_SERVICE_ACCOUNT_PATH"))
firebase_admin.initialize_app(cred)
sg = SendGridAPIClient(os.getenv("SENDGRID_API_KEY"))
FROM_EMAIL = os.getenv("SENDGRID_FROM_EMAIL")

# 存储上次检查的时间（可以写入本地文件持久化，避免重启后丢失）
last_check_time = datetime.now(timezone.utc)

def send_welcome_email(to_email):
    """同方案1的发送函数"""
    message = Mail(
        from_email=FROM_EMAIL,
        to_emails=to_email,
        subject="🎉 欢迎加入我们的平台！",
        html_content="<strong>很高兴你能来！</strong><p>感谢你注册我们的应用，期待你的精彩使用体验。</p>"
    )
    try:
        response = sg.send(message)
        print(f"✅ 欢迎邮件已发送到 {to_email}，状态码: {response.status_code}")
    except Exception as e:
        print(f"❌ 发送邮件失败 ({to_email}): {str(e)}")

def check_for_new_users():
    global last_check_time
    print(f"🔍 开始检查新用户，上次检查时间: {last_check_time.strftime('%Y-%m-%d %H:%M:%S UTC')}")
    
    # 分页获取Firebase Auth用户（一次最多返回1000条）
    page = auth.list_users()
    while page:
        for user in page.users:
            # 转换用户创建时间为UTC datetime
            create_timestamp = user.user_metadata.creation_timestamp
            create_datetime = datetime.fromtimestamp(create_timestamp / 1000, timezone.utc)
            
            if create_datetime > last_check_time:
                send_welcome_email(user.email)
        
        # 获取下一页用户
        page = page.get_next_page()
    
    # 更新上次检查时间
    last_check_time = datetime.now(timezone.utc)
    print("✅ 新用户检查完成")

if __name__ == "__main__":
    print("🚀 定时新用户检查服务已启动，每5分钟检查一次...")
    while True:
        check_for_new_users()
        time.sleep(300)  # 5分钟检查一次，可根据需求调整间隔