Django Channels WebSocket消费者阻止同步API更新数据库问题

问题现象 #

• WebSocket连接ChatConsumer时，调用SessionTakeOverAPIView无法更新数据库；断开连接后API功能恢复正常。
• 通过Django Admin更新Session数据，即使消费者处于连接状态也能正常生效。
• 移除消费者中self.org = self.session.org的赋值后，API可正常更新数据库；移除API中发送消息到channel_layer的代码后，API也能正常工作。
• 尝试使用后台任务解决该问题，但未生效。

相关代码 #

SessionTakeOverAPIView（同步API） #

class SessionTakeOverAPIView(generics.GenericAPIView):
    """
    该API视图支持人工或AI接管聊天会话，处理会话接管验证、更新会话状态，并向聊天组广播相关事件。
    使用POST请求触发人工或AI接管会话，访问需要身份验证。
    """

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.room_group_name = None

    permission_classes = [BotUserHasRequiredPermissionForMethod]
    post_permission_required = ['session.reply_session']
    queryset = Session.objects.select_related('bot').all()
    serializer_class = SessionTakeOverSerializer

    def get_object(self):
        """
        根据请求数据中的session_id获取会话对象，会话不存在时返回404错误。
        """
        try:
            return super().get_queryset().get(session_id=self.request.data.get('session_id'))
        except Session.DoesNotExist:
            raise Http404  # 会话不存在时返回404错误


    def handle_human_take_over(self):
        """
        处理人工接管聊天会话的逻辑：
        - 验证会话是否已被其他人接管
        - 更新会话状态，标记当前用户为接管人
        - 向聊天组发送接管通知消息
        - 异步创建接管日志记录
        """
        request = self.request
        session: Session = self.get_object()

        # 检查会话是否已被其他人工接管
        if session.is_session_currently_taken_over:
            raise ValidationError({
                "detail" :f"会话当前已被{session.currently_taken_over_by.get_full_name}接管"})

        # 准备接管通知消息
        take_over_msg = f"{session.bot.human_take_over_msg} {request.user.get_full_name}"

        # 准备发送到组的响应消息
        response = {
            "event_type": EventTypeChoicesChat.HUMAN_TAKEOVER,
            "user_type": "client",
            "sender_icon": request.user.profile_icon_url,

            "chat_data": {
                "msg": take_over_msg,
                "recorded_audio_msg_link": None,
                "images_link": [],
                "files_link": []
            }
        }

        # 更新会话状态以反映人工接管
        session.currently_taken_over_by = request.user
        session.is_session_currently_taken_over = True
        session.is_ai_active = False  # AI不再控制会话

        session.save(
            update_fields=[
                'currently_taken_over_by',
                'is_session_currently_taken_over',
                'is_ai_active'
            ]
        )

        transaction.on_commit(
            lambda: send_data_to_channel_layer(
                self.room_group_name, 
                'send_chat_message', 
                response
            )
        )

        # 异步创建会话接管操作日志
        create_session_take_over_info.delay(session.id, request.user.id)

    def handle_ai_take_over(self):
        """
        处理AI接管聊天会话的逻辑：
        - 验证AI是否已在会话中激活
        - 更新会话状态以反映AI控制，并向聊天组发送通知消息
        """
        session: Session = self.get_object()

        # 检查AI是否已在会话中激活
        if session.is_ai_active:
            raise ValidationError({"detail": "AI已在运行中"})

        # 准备AI接管通知消息
        take_over_msg = session.bot.ai_take_over_msg

        # 准备发送到聊天室组的响应消息
        response = {
            "event_type": EventTypeChoicesChat.AI_TAKE_OVER,
            "user_type": "ai",
            "sender_icon": session.bot.bot_icon,

            "chat_data": {
                "msg": take_over_msg,
                "recorded_audio_msg_link": None,
                "images_link": [],
                "files_link": []
            }
        }

        # 更新会话状态以反映AI接管
        session.currently_taken_over_by = None  # 无人工控制
        session.is_session_currently_taken_over = False
        session.is_ai_active = True  # AI现在控制会话
        
        session.save(
            update_fields=[
                'currently_taken_over_by',
                'is_session_currently_taken_over',
                'is_ai_active'
            ]
        )

        transaction.on_commit(
            lambda: send_data_to_channel_layer(
                self.room_group_name, 
                'send_chat_message', 
                response
            )
        )


    def post(self, request, *args, **kwargs):
        """
        处理POST请求以接管会话（人工或AI）：
        验证会话ID，并检查会话是否处于激活状态，然后允许人工或AI接管。
        """

        serializer = self.serializer_class(data=request.data)
        serializer.is_valid(raise_exception=True)

        # 根据请求数据确定接管类型（人工或AI）
        take_over_type = serializer.validated_data.get('take_over_type')
        self.room_group_name = constants.CHAT_SESSION_ROOM_GROUP_NAME.format(
            session_id=request.data.get('session_id')
        )  

        session: Session = self.get_object()
        if session.session_has_ended:
            raise ValidationError({"detail" :"会话已失效。"})

        with transaction.atomic():
            if take_over_type == EventTypeChoicesChat.HUMAN_TAKEOVER:
                self.handle_human_take_over()  # 处理人工接管
            elif take_over_type == EventTypeChoicesChat.AI_TAKE_OVER:
                self.handle_ai_take_over()  # 处理AI接管
            else:
                raise ValidationError({"detail": "无效的接管类型。"})

        return Response({"message": "接管成功"})

ChatConsumer（异步WebSocket消费者） #

class ChatConsumer(AsyncWebsocketConsumer):
    """ 处理特定会话中聊天消息的WebSocket消费者 """

    async def connect(self):
        """处理WebSocket连接初始化"""
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = constants.CHAT_SESSION_ROOM_GROUP_NAME.format(session_id=self.room_name)

        session = await self.__get_session({"session_id": self.room_name, "session_has_ended": False})
        if not session:
            logger.info(f"会话 [{self.room_name}] 已过期或不存在")
            # 关闭连接
            await self.close()
            return

        self.session = session

        await self.__set_attributes()
        await self.channel_layer.group_add(self.room_group_name, self.channel_name)
        await self.accept()

        # 验证连接条件
        if not await self.__validate_connection():
            await self.__send_validation_failed_message()
            return

        # 如果没有消息则发送欢迎消息
        if not await database_sync_to_async(lambda: session.has_messages)():
            await self.__send_intro_message()


    async def __set_attributes(self):
        self.org = self.session.org
        self.bot = self.session.bot
        self.platform = self.bot.platform
        self.language = self.bot.language
        self.ai_response_url = settings.AI_BASE_URL + f'/api/v1.0/ai/conversation/{self.org.id}/{self.bot.id}'
        self.goal = '你是一名智能客服代理'
        self.customer_name = self.session.user_name if self.session.user_name else ""
        self.customer_email = self.session.user_email if self.session.user_email else ""
        self.user_type = None
        self.user = self.scope["user"] if not self.scope["user"].is_anonymous else None


    @database_sync_to_async
    def __get_session(self, filter_kwargs: dict):
        """获取包含关联数据的会话"""
        try:
            session = Session.objects.select_related(
                "bot",
                "org",
                "bot__language", 
                "bot__platform"
            ).get(**filter_kwargs)
            return session
        except Exception as err:
            logger.error(f"获取会话时出错: {err}")
            return None

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