Python處理application/json錯誤的方法詳解

更新時間：2025年02月24日 09:22:50 作者：Maybe_9527

這篇文章主要為大家詳細介紹了python使用httpx_sse調(diào)用sse流式接口對響應格式為application/json的錯誤信息處理的相關知識,需要的可以了解下

問題描述

調(diào)用sse流式接口使用httpx_sse的方式

				import httpx
				from httpx_sse import connect_sse
				# 省略無關代碼
				try:
                    with httpx.Client() as client:
                        with connect_sse(client, "GET", url, params=param) as event_source:
                            clear_textbox(response_textbox)
                            # 把 iter_sse() 迭代完, 就相當于處理完了一次流式調(diào)用
                            for sse in event_source.iter_sse():
                                # 流式響應中,每次響應體的處理邏輯
                                print(f"generated_answer的值是: '{sse.data}'")
                                response = sse.data
                                if response != '':
                                    # self.response = response
                                    append_text(response_textbox, response)

                except httpx.RequestError as e:
                    print(f"請求錯誤：{e}")
                except Exception as e:
                    print(f"發(fā)生了一個錯誤：{e}")

httpx_sse的connet_sse源碼：

@contextmanager
def connect_sse(
    client: httpx.Client, method: str, url: str, **kwargs: Any
) -> Iterator[EventSource]:
    headers = kwargs.pop("headers", {})
    headers["Accept"] = "text/event-stream"
    headers["Cache-Control"] = "no-store"

    with client.stream(method, url, headers=headers, **kwargs) as response:
        yield EventSource(response)

可以看到connect_sse源碼中的headers的"Accept"設置了只接受"text/event-stream"流式結果，正常這么調(diào)用是沒錯的。但是當后端的流式接口因為401權限問題等報錯返回了"application/json"格式，如
{ “code”:401, “msg”:“登錄過期，請重新登錄”, “data”:null} 這樣的json格式結果時，以上代碼就會報錯，因為他不是"text/event-stream"流式響應結果頭。那么該怎么辦呢？

方案

重新寫一個自定義的connect_sse。

import httpx
from httpx_sse import EventSource
from typing import Any, Iterator
from contextlib import contextmanager
import json
# 自定義調(diào)用sse接口
    @contextmanager
    def custom_connect_sse(
            self, client: httpx.Client, method: str, url: str, **kwargs: Any
    ) -> Iterator[EventSource]:
        headers = kwargs.pop("headers", {})
        # 只有當沒有指定Accept時才添加默認值
        headers["Accept"] = "*/*"
        headers["Cache-Control"] = "no-store"

        with client.stream(method, url, headers=headers, **kwargs) as response:
            content_type = response.headers.get('content-type', '').lower()
            json_flag = False
            if 'text/event-stream' in content_type:
                # 處理SSE流
                yield json_flag, EventSource(response)
            elif 'application/json' in content_type:
                # yield response  # 在這里你可以決定如何進一步處理這個JSON響應
                # 讀取并合并所有文本塊
                text_data = ''.join([chunk for chunk in response.iter_text()])
                # 解析整個響應體為JSON
                json_data = json.loads(text_data)
                json_flag = True
                yield json_flag, json_data

調(diào)用代碼

# 使用自定義的connect_sse函數(shù)
                try:
                    with httpx.Client() as client:
                        with self.custom_connect_sse(client, "GET", url, params=param, headers=headers) as (json_flag, event_source):
                            if json_flag:
                                code = event_source.get("code")
                                msg = event_source.get("msg")
                                print(f"Code: [code], Message: {msg}")
                            else:
                                full_answer = ""
                                clear_textbox(response_textbox)
                                for sse in event_source.iter_sse():
                                    print(f"generated_answer的值是: '{sse.data}'")
                                    response = sse.data
                                    if response:
                                        append_text(response_textbox, response)
                                        full_answer += response
                                user_record += reply + full_answer + "\n"
                                print(f"user_record:{user_record}")

                except httpx.RequestError as e:
                    print(f"請求錯誤：{e}")
                except Exception as e:
                    print(f"發(fā)生了一個錯誤：{e}")

關鍵步驟：

1.設置headers[“Accept”] = “/”，所有響應頭都可以接收

2.content_type = response.headers.get(‘content-type’, ‘’).lower() 判斷響應頭是流式還是json，并用json_flag記錄是否json標識，返回不同的結果。如果是json，則循環(huán)合并處理chunk塊，拼裝完整json返回結果（實測第一次就返回完整json結構了，但是代碼得這么寫）。

3.使用自定義connect_sse方法時，根據(jù)json_flag來分別處理成功調(diào)用流式結果還是異常的json結果。

到此這篇關于Python處理application/json錯誤的方法詳解的文章就介紹到這了,更多相關Python處理application/json錯誤內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: