""" Response — 兼容 requests.Response 的响应对象 属性/方法与 requests.Response 一致: .status_code .text .content .headers .ok .url .encoding .json() .raise_for_status() .iter_content() """ import json as _json class Response: """兼容 requests.Response 的响应对象。 额外属性: session_id: Bluebird 返回的 sessionId raw: Bluebird 原始响应 dict """ def __init__(self, status_code=0, text="", headers=None, url="", raw=None): self.status_code = int(status_code) if status_code else 0 self._text = text or "" self.headers = dict(headers) if headers else {} self.url = url self.encoding = "utf-8" self.raw = raw or {} self.content = self._text.encode(self.encoding) if self._text else b"" self.ok = 200 <= self.status_code < 400 self.cookies = {} @property def text(self): return self._text def json(self, **kwargs): return _json.loads(self._text, **kwargs) def raise_for_status(self): """status_code >= 400 时抛出 HTTPError。""" if not self.ok: from .exceptions import HTTPError raise HTTPError(self) def iter_content(self, chunk_size=1): """兼容 requests.Response.iter_content()。""" if self.content: for i in range(0, len(self.content), chunk_size): yield self.content[i:i + chunk_size] @property def session_id(self): return self.raw.get("sessionId", "") @property def reason(self): """兼容 requests.Response.reason。""" reasons = { 200: "OK", 201: "Created", 202: "Accepted", 301: "Moved Permanently", 302: "Found", 400: "Bad Request", 401: "Unauthorized", 403: "Forbidden", 404: "Not Found", 500: "Internal Server Error", } return reasons.get(self.status_code, "Unknown") def __repr__(self): return f"" def __bool__(self): return self.ok def __len__(self): return len(self.content)