(self, code, obj)
| 733 | pass |
| 734 | |
| 735 | def _send_json(self, code, obj): |
| 736 | body = json.dumps(obj, ensure_ascii=False).encode() |
| 737 | self.send_response(code) |
| 738 | self.send_header("Content-Type", "application/json") |
| 739 | self.send_header("Content-Length", str(len(body))) |
| 740 | if self.close_connection: |
| 741 | # 主动关闭连接时显式告知客户端,避免其在已关闭的 socket 上复用连接。 |
| 742 | self.send_header("Connection", "close") |
| 743 | self.end_headers() |
| 744 | self.wfile.write(body) |
| 745 | |
| 746 | def _sse(self, event, data): |
| 747 | chunk = f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode() |
no outgoing calls
no test coverage detected