Detailed explanation of Python's method of handling application/json errors

Problem description

How to use httpx_sse when calling the sse streaming interface

				import httpx
				from httpx_sse import connect_sse
				# Omit irrelevant code				try:
                    with () as client:
                        with connect_sse(client, "GET", url, params=param) as event_source:
                            clear_textbox(response_textbox)
                            # Iterate iter_sse(), which is equivalent to processing a streaming call                            for sse in event_source.iter_sse():
                                # In streaming response, the processing logic of each response body                                print(f"generated_answerThe value is: '{}'")
                                response = 
                                if response != '':
                                    #  = response
                                    append_text(response_textbox, response)

                except  as e:
                    print(f"Request error：{e}")
                except Exception as e:
                    print(f"An error occurred：{e}")

Connet_sse source code of httpx_sse:

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

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

You can see that the "Accept" of headers in the connect_sse source code sets to only accept "text/event-stream" streaming results. It is correct to call this normally. However, when the backend streaming interface reports an error due to 401 permissions, etc., it returns the "application/json" format, such as
{ "code":401, "msg": "Login expires, please log in again", "data":null}, the above code will report an error because it is not the "text/event-stream" streaming response result header. So what should I do?

plan

Rewrite a custom connect_sse.

import httpx
from httpx_sse import EventSource
from typing import Any, Iterator
from contextlib import contextmanager
import json
# Customize the call to the sse interface    @contextmanager
    def custom_connect_sse(
            self, client: , method: str, url: str, **kwargs: Any
    ) -&gt; Iterator[EventSource]:
        headers = ("headers", {})
        # Add default value only if Accept is not specified        headers["Accept"] = "*/*"
        headers["Cache-Control"] = "no-store"

        with (method, url, headers=headers, **kwargs) as response:
            content_type = ('content-type', '').lower()
            json_flag = False
            if 'text/event-stream' in content_type:
                # Process SSE flow                yield json_flag, EventSource(response)
            elif 'application/json' in content_type:
                # yield response # Here you can decide how to further handle this JSON response                # Read and merge all text blocks                text_data = ''.join([chunk for chunk in response.iter_text()])
                # parse the entire response body to JSON                json_data = (text_data)
                json_flag = True
                yield json_flag, json_data

Call code

# Use the custom connect_sse function                try:
                    with () 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_answerThe value is: '{}'")
                                    response = 
                                    if response:
                                        append_text(response_textbox, response)
                                        full_answer += response
                                user_record += reply + full_answer + "\n"
                                print(f"user_record:{user_record}")

                except  as e:
                    print(f"Request error：{e}")
                except Exception as e:
                    print(f"An error occurred：{e}")

Key steps:

1. Set headers["Accept"] = "/", all response headers can be received

2.content_type = (‘content-type’, ‘’).lower() determines whether the response header is streaming or json, and uses json_flag to record whether the json is identified, returning different results. If it is json, then loop merge and process the chunk block, assemble the complete json and return the result (the actual test returns the complete json structure for the first time, but the code must be written like this).

3. When using the custom connect_sse method, the json_flag is used to process the result of successful call to stream or exception json results.

This is the end of this article about the detailed explanation of Python's method of handling application/json errors. For more related content on Python's application/json errors, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!