-
Notifications
You must be signed in to change notification settings - Fork 0
Open
Description
The Request and Response classes should ensure that their accessor properties are ⛔read-only properties⛔ rather than settable attributes.
Use cases that require an altered version of a request or response should explicitly create new copies.
Clone a request, switching the URL scheme to https...
req = Request(
method=req.method,
url=req.url.copy_with(scheme="https"),
headers=req.headers,
content=req.stream,
)Clone a response, applying a gzip encoding...
res = Response(
status=res.status,
headers=res.headers.copy_set("Content-Encoding", "gzip"),
content=GZipStream(res.stream),
)We should document the patterns for copying requests/responses. 👍🏼
Copying instances like this is also useful if users want to use custom response classes. 💭
A custom response class...
class APIResponse(Response):
def __init__(self, status_code, headers, content):
self.__super__(status_code, headers, content)
def parse(self):
status_code = self.status_code
content_type = self.headers.get("Content-Type")
if status_code != 200:
raise ValueError(r"Expected code 200, got {status_code}")
if content_type != "application/json":
raise ValueError(r"Expected "application/json", got {content_type}")
self.read()
self._data = json.loads(self.body)
@property
def data(self):
if not hasattr(self, '_data'):
raise RuntimeError("Cannot access `.data` without first calling `.parse()`")
return self._dataA custom client that returns custom responses...
class APIClient:
def __init__(self):
self.url = httpx.URL('https://www.example.com')
self.headers = httpx.Headers({
'Accept-Encoding': 'gzip',
'Connection': 'keep-alive',
'User-Agent': 'dev'
})
self.via = httpx.RedirectMiddleware(httpx.ConnectionPool())
def get(self, path: str) -> Response:
req = httpx.Request(
method="GET",
url=self.url.join(path),
headers=self.headers,
)
with self.via.send(req) as resp:
response = APIResponse(
resp.status_code,
headers=resp.headers,
content=resp.stream)
)
response.parse()
return response
def post(self, path: str, payload: Any) -> httpx.Response:
req = httpx.Request(
method="POST",
url=self.url.join(path),
headers=self.headers,
content=httpx.JSON(payload),
)
with self.via.send(req) as resp:
response = APIResponse(
resp.status_code,
headers=resp.headers,
content=resp.stream)
)
response.parse()
return responseReactions are currently unavailable
Metadata
Metadata
Assignees
Labels
No labels