如何处理GCP Marketplace Pub/Sub订阅事件以实现SaaS集成？

基于你提供的Pub/Sub消息处理代码，以下是不同系统架构下的用户审批及权限配置的实现片段：

1. Firebase Auth 用户状态同步（扩展现有示例） #

如果使用Firebase Auth管理用户身份，可以同步更新用户自定义声明，实现权限控制：

import firebase_admin
from firebase_admin import auth
from firebase_admin import firestore

# 初始化Firebase Admin（确保项目已完成初始化）
if not firebase_admin._apps:
    firebase_admin.initialize_app()
db = firestore.client()

def approve_account_with_firebase_auth(account_id, entitlements):
    try:
        # 更新用户自定义声明，添加权限标识
        auth.set_custom_user_claims(account_id, {
            "approved": True,
            "roles": entitlements  # 示例：["admin", "editor"]
        })
        print(f"用户 {account_id} 的Auth声明已更新，权限：{entitlements}")
        
        # 同步更新Firestore用户状态（结合原有逻辑）
        db.collection("accounts").document(account_id).set({
            "status": "approved",
            "entitlements": entitlements,
            "updated_at": firestore.SERVER_TIMESTAMP
        }, merge=True)
    except auth.UserNotFoundError:
        print(f"用户 {account_id} 在Firebase Auth中不存在")
    except Exception as e:
        print(f"更新用户权限失败：{str(e)}")

# 在handle_pubsub_event函数中替换原有Firebase代码：
if account_id:
    approve_account_with_firebase_auth(account_id, entitlements)

2. 传统SQL数据库（以PostgreSQL为例）的用户审批 #

如果使用SQL数据库存储用户信息，可通过更新用户表状态和关联权限表实现：

import psycopg2
from psycopg2.extras import DictCursor

def approve_account_with_sql(account_id, entitlements):
    conn = None
    try:
        # 建立数据库连接
        conn = psycopg2.connect(
            dbname="your_db_name",
            user="your_db_user",
            password="your_db_pass",
            host="localhost"
        )
        cur = conn.cursor(cursor_factory=DictCursor)
        
        # 1. 更新用户状态为已审批
        cur.execute("""
            UPDATE users 
            SET status = %s, updated_at = CURRENT_TIMESTAMP
            WHERE id = %s
        """, ("approved", account_id))
        
        # 2. 清理原有权限并插入新权限（避免重复）
        cur.execute("DELETE FROM user_entitlements WHERE user_id = %s", (account_id,))
        for role in entitlements:
            cur.execute("""
                INSERT INTO user_entitlements (user_id, entitlement)
                VALUES (%s, %s)
            """, (account_id, role))
        
        conn.commit()
        print(f"用户 {account_id} 已审批，权限已更新")
    except psycopg2.Error as e:
        if conn:
            conn.rollback()
        print(f"数据库操作失败：{str(e)}")
    finally:
        if conn:
            cur.close()
            conn.close()

# 在Pub/Sub处理函数中调用
if account_id:
    approve_account_with_sql(account_id, entitlements)

3. RBAC（基于角色的访问控制）权限模型实现 #

如果需要更细粒度的权限控制，可实现RBAC逻辑：

from firebase_admin import firestore

db = firestore.client()

# 角色-权限映射（可存储在数据库或配置文件中）
ROLE_PERMISSIONS = {
    "admin": ["create_user", "delete_user", "view_all_data"],
    "editor": ["edit_data", "view_data"],
    "viewer": ["view_data"]
}

def get_permissions_from_roles(roles):
    """根据角色列表获取完整权限集合"""
    permissions = set()
    for role in roles:
        permissions.update(ROLE_PERMISSIONS.get(role, []))
    return list(permissions)

def approve_account_with_rbac(account_id, roles):
    # 获取用户完整权限列表
    permissions = get_permissions_from_roles(roles)
    
    # 将角色和权限存入数据库
    db.collection("accounts").document(account_id).set({
        "status": "approved",
        "roles": roles,
        "permissions": permissions,
        "updated_at": firestore.SERVER_TIMESTAMP
    }, merge=True)
    print(f"用户 {account_id} 已审批，角色：{roles}，权限：{permissions}")

# 使用示例
if account_id:
    approve_account_with_rbac(account_id, entitlements)

额外的最佳实践 #

• 幂等性处理：Pub/Sub可能重复推送消息，添加消息ID去重逻辑避免重复审批

  # 在handle_pubsub_event中加入去重逻辑
  message_id = pubsub_message.get('messageId')
  if db.collection("processed_messages").document(message_id).get().exists:
      return "Message already processed", 200
  # 处理完成后记录已处理的消息ID
  db.collection("processed_messages").document(message_id).set({"processed_at": firestore.SERVER_TIMESTAMP})
  

• 日志与监控：记录审批操作的详细日志，便于问题追踪
• 权限校验：在业务接口中添加权限验证逻辑，确保只有已审批且拥有对应权限的用户可访问资源

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