如何在Python+FastAPI中实现跨双服务的事务性删除？

由于两个数据库属于独立系统，无法使用传统单机ACID事务保证强一致性，以下是几种可行的Python实现方案，结合你的FastAPI代码场景优化：

一、补偿事务（TCC）模式 #

核心思路是先执行可逆操作，确认所有步骤成功后再提交不可逆操作，失败则回滚已完成步骤，适合同步场景：

方案1：状态标记式补偿 #

将自有数据库的删除操作拆分为「标记待删除」→「确认删除」两步，失败则取消标记：

async def delete_db_firebase(self, *, _id: Union[int, str], route: Optional[str] = "") -> Any:
    url_database = f"{self.url}{route}/{_id}"
    url_firebase = f"{settings.AT_PRONOSTICOS_AUTH}/api/users/{_id}"
    
    try:
        # 1. 标记自有数据库用户为待删除（假设自有库支持状态更新接口）
        mark_resp = await self._client.patch(
            url_service=url_database,
            json={"status": "pending_deletion"}
        )
        await self._check_codes.check_codes(response=mark_resp)
        
        # 2. 删除Firebase用户
        firebase_resp = await self._client.delete(url_service=url_firebase)
        await self._check_codes.check_codes(response=firebase_resp, delete_method=True)
        
        # 3. 确认删除自有数据库用户
        db_resp = await self._client.delete(url_service=url_database)
        await self._check_codes.check_codes(response=db_resp, delete_method=True)
        
        return db_resp
    except Exception as e:
        # 补偿：取消自有数据库用户的待删除标记
        try:
            await self._client.patch(
                url_service=url_database,
                json={"status": "active"}
            )
        except Exception as rollback_err:
            self._logger.error(f"回滚用户{_id}状态失败: {str(rollback_err)}")
        raise e

方案2：备份恢复式补偿 #

如果自有库不支持状态标记，先备份用户数据再执行删除，失败则从备份恢复：

async def delete_db_firebase(self, *, _id: Union[int, str], route: Optional[str] = "") -> Any:
    url_database = f"{self.url}{route}/{_id}"
    url_firebase = f"{settings.AT_PRONOSTICOS_AUTH}/api/users/{_id}"
    backup_url = f"{self.url}/backup_users/{_id}"
    
    try:
        # 1. 备份自有数据库用户到临时表
        user_data_resp = await self._client.get(url_service=url_database)
        await self._check_codes.check_codes(response=user_data_resp)
        await self._client.post(url_service=backup_url, json=user_data_resp.json())
        
        # 2. 删除自有数据库用户
        db_resp = await self._client.delete(url_service=url_database)
        await self._check_codes.check_codes(response=db_resp, delete_method=True)
        
        # 3. 删除Firebase用户
        firebase_resp = await self._client.delete(url_service=url_firebase)
        await self._check_codes.check_codes(response=firebase_resp, delete_method=True)
        
        # 4. 删除备份数据
        await self._client.delete(url_service=backup_url)
        return db_resp
    except Exception as e:
        # 补偿：从备份恢复自有数据库用户
        try:
            backup_resp = await self._client.get(url_service=backup_url)
            await self._check_codes.check_codes(response=backup_resp)
            await self._client.post(url_service=url_database, json=backup_resp.json())
            await self._client.delete(url_service=backup_url)
        except Exception as rollback_err:
            self._logger.error(f"恢复用户{_id}失败: {str(rollback_err)}")
        raise e

二、幂等性重试+回滚 #

如果删除操作本身是幂等的（比如Firebase删除不存在的用户不报错），可以通过重试失败步骤+回滚已完成操作保证一致性：

from tenacity import retry, stop_after_attempt, wait_exponential  # 需安装tenacity库

async def restore_user(self, *, _id: Union[int, str], route: Optional[str] = "", user_data: dict) -> None:
    # 恢复自有数据库用户的方法
    url_database = f"{self.url}{route}/{_id}"
    restore_resp = await self._client.post(url_service=url_database, json=user_data)
    await self._check_codes.check_codes(response=restore_resp)

async def delete_db_firebase(self, *, _id: Union[int, str], route: Optional[str] = "") -> Any:
    url_database = f"{self.url}{route}/{_id}"
    url_firebase = f"{settings.AT_PRONOSTICOS_AUTH}/api/users/{_id}"
    
    # 提前获取用户数据用于回滚
    user_data_resp = await self._client.get(url_service=url_database)
    await self._check_codes.check_codes(response=user_data_resp)
    user_data = user_data_resp.json()
    
    try:
        # 删除自有数据库用户
        db_resp = await self._client.delete(url_service=url_database)
        await self._check_codes.check_codes(response=db_resp, delete_method=True)
        
        # 重试删除Firebase用户（最多3次）
        @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
        async def delete_firebase():
            resp = await self._client.delete(url_service=url_firebase)
            await self._check_codes.check_codes(response=resp, delete_method=True)
            return resp
        
        await delete_firebase()
        return db_resp
    except Exception as e:
        # 回滚自有数据库用户
        try:
            await self.restore_user(_id=_id, route=route, user_data=user_data)
        except Exception as rollback_err:
            self._logger.error(f"恢复用户{_id}失败: {str(rollback_err)}")
        raise e

三、异步最终一致性（消息队列） #

如果业务允许异步处理，可通过消息队列（如Redis Queue、Celery）实现最终一致性：

1. 接口收到删除请求后，向队列发送「删除用户」任务
2. 消费任务时，先删除自有数据库用户
3. 自有库删除成功后，再删除Firebase用户
4. 某一步失败时，队列自动重试，或触发人工干预

这种方式避免接口阻塞，适合实时性要求不高的场景。

关键注意事项 #

• 幂等性保障：确保删除接口重复执行无副作用（比如Firebase删除不存在用户返回成功状态）
• 日志记录：所有操作和异常需详细记录，方便后续排查和人工补偿
• 人工兜底：分布式事务无完美解决方案，需预留人工处理不一致数据的入口

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