Add Response.raise_for_excepted_status() method
Add a new method that raises HTTPStatusError unless the status code is explicitly listed in the expected parameter. Unlike `raise_for_status()`, this method requires all acceptable status codes (including 2xx) to be explicitly specified. Also refactors shared logic into `_ensure_request()` and `_raise_status_error()` helper methods.
This commit is contained in:
parent
ae1b9f6623
commit
3583fa7465
@ -71,6 +71,7 @@
|
||||
[total_seconds()](https://docs.python.org/3/library/datetime.html#datetime.timedelta.total_seconds) to correctly get
|
||||
the total elapsed seconds.
|
||||
* `def .raise_for_status()` - **Response**
|
||||
* `def .raise_for_excepted_status(expected)` - **Response**
|
||||
* `def .json()` - **Any**
|
||||
* `def .read()` - **bytes**
|
||||
* `def .iter_raw([chunk_size])` - **bytes iterator**
|
||||
|
||||
@ -305,6 +305,30 @@ The method returns the response instance, allowing you to use it inline. For exa
|
||||
>>> data = httpx.get('...').raise_for_status().json()
|
||||
```
|
||||
|
||||
### Allowing Specific Status Codes
|
||||
|
||||
Sometimes you may expect certain non-2xx status codes as valid responses (e.g., 404 when checking if a resource exists). Use `raise_for_excepted_status()` to specify which status codes are acceptable:
|
||||
|
||||
```pycon
|
||||
>>> r = httpx.get('https://httpbin.org/status/404')
|
||||
>>> r.raise_for_excepted_status([200, 404]) # 404 is expected, no exception raised
|
||||
<Response [404 Not Found]>
|
||||
```
|
||||
|
||||
Note that `raise_for_excepted_status()` only allows the status codes explicitly listed in the `expected` parameter. Even 2xx success codes must be included:
|
||||
|
||||
```pycon
|
||||
>>> r = httpx.get('https://httpbin.org/get')
|
||||
>>> r.status_code
|
||||
200
|
||||
>>> r.raise_for_excepted_status([201]) # 200 not in list, raises exception
|
||||
Traceback (most recent call last):
|
||||
...
|
||||
httpx._exceptions.HTTPStatusError: ...
|
||||
>>> r.raise_for_excepted_status([200, 201]) # 200 is in list, passes
|
||||
<Response [200 OK]>
|
||||
```
|
||||
|
||||
## Response Headers
|
||||
|
||||
The response headers are available as a dictionary-like interface.
|
||||
|
||||
@ -791,20 +791,19 @@ class Response:
|
||||
and "Location" in self.headers
|
||||
)
|
||||
|
||||
def raise_for_status(self) -> Response:
|
||||
"""
|
||||
Raise the `HTTPStatusError` if one occurred.
|
||||
"""
|
||||
request = self._request
|
||||
if request is None:
|
||||
def _ensure_request(self, method_name: str) -> Request:
|
||||
"""Ensure request is set, raise RuntimeError if not."""
|
||||
if self._request is None:
|
||||
raise RuntimeError(
|
||||
"Cannot call `raise_for_status` as the request "
|
||||
f"Cannot call `{method_name}` as the request "
|
||||
"instance has not been set on this response."
|
||||
)
|
||||
return self._request
|
||||
|
||||
if self.is_success:
|
||||
return self
|
||||
|
||||
def _raise_status_error(
|
||||
self, request: Request, *, error_type_for_2xx: str | None = None
|
||||
) -> typing.NoReturn:
|
||||
"""Internal helper to raise HTTPStatusError with appropriate message."""
|
||||
if self.has_redirect_location:
|
||||
message = (
|
||||
"{error_type} '{0.status_code} {0.reason_phrase}' for url '{0.url}'\n"
|
||||
@ -818,16 +817,58 @@ class Response:
|
||||
)
|
||||
|
||||
status_class = self.status_code // 100
|
||||
error_types = {
|
||||
error_types: dict[int, str] = {
|
||||
1: "Informational response",
|
||||
3: "Redirect response",
|
||||
4: "Client error",
|
||||
5: "Server error",
|
||||
}
|
||||
if error_type_for_2xx is not None:
|
||||
error_types[2] = error_type_for_2xx
|
||||
|
||||
error_type = error_types.get(status_class, "Invalid status code")
|
||||
message = message.format(self, error_type=error_type)
|
||||
raise HTTPStatusError(message, request=request, response=self)
|
||||
|
||||
def raise_for_status(self) -> Response:
|
||||
"""
|
||||
Raise the `HTTPStatusError` if one occurred.
|
||||
"""
|
||||
request = self._ensure_request("raise_for_status")
|
||||
|
||||
if self.is_success:
|
||||
return self
|
||||
|
||||
self._raise_status_error(request)
|
||||
|
||||
def raise_for_excepted_status(
|
||||
self, expected: typing.Sequence[int]
|
||||
) -> Response:
|
||||
"""
|
||||
Raise the `HTTPStatusError` unless the status code is in the `expected` list.
|
||||
|
||||
Only status codes explicitly listed in `expected` are allowed to pass.
|
||||
All other status codes (including 2xx) will raise an exception.
|
||||
|
||||
Args:
|
||||
expected: A sequence of status codes that are considered acceptable
|
||||
and should not raise an exception.
|
||||
|
||||
Returns:
|
||||
This response instance if the status code is in the expected list.
|
||||
|
||||
Raises:
|
||||
HTTPStatusError: If the response status code is not in the expected list.
|
||||
"""
|
||||
request = self._ensure_request("raise_for_excepted_status")
|
||||
|
||||
if self.status_code in expected:
|
||||
return self
|
||||
|
||||
self._raise_status_error(
|
||||
request, error_type_for_2xx="Unexpected success response"
|
||||
)
|
||||
|
||||
def json(self, **kwargs: typing.Any) -> typing.Any:
|
||||
return jsonlib.loads(self.content, **kwargs)
|
||||
|
||||
|
||||
@ -146,6 +146,64 @@ def test_raise_for_status():
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
def test_raise_for_excepted_status():
|
||||
request = httpx.Request("GET", "https://example.org")
|
||||
|
||||
# 2xx status code in expected list - should pass
|
||||
response = httpx.Response(200, request=request)
|
||||
assert response.raise_for_excepted_status([200]) is response
|
||||
|
||||
# 2xx status code NOT in expected list - should raise with "Unexpected success"
|
||||
response = httpx.Response(200, request=request)
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
response.raise_for_excepted_status([201, 204])
|
||||
assert "Unexpected success response '200 OK'" in str(exc_info.value)
|
||||
|
||||
# 4xx status code in expected list - should pass
|
||||
response = httpx.Response(404, request=request)
|
||||
assert response.raise_for_excepted_status([200, 404]) is response
|
||||
|
||||
# 4xx status code NOT in expected list - should raise
|
||||
response = httpx.Response(404, request=request)
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
response.raise_for_excepted_status([200, 400])
|
||||
assert "Client error '404 Not Found'" in str(exc_info.value)
|
||||
|
||||
# 5xx status code in expected list - should pass
|
||||
response = httpx.Response(500, request=request)
|
||||
assert response.raise_for_excepted_status([500, 502, 503]) is response
|
||||
|
||||
# 5xx status code NOT in expected list - should raise
|
||||
response = httpx.Response(500, request=request)
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
response.raise_for_excepted_status([200])
|
||||
assert "Server error '500 Internal Server Error'" in str(exc_info.value)
|
||||
|
||||
# 3xx redirect in expected list - should pass
|
||||
headers = {"location": "https://other.org"}
|
||||
response = httpx.Response(301, headers=headers, request=request)
|
||||
assert response.raise_for_excepted_status([301, 302]) is response
|
||||
|
||||
# 3xx redirect NOT in expected list - should raise with redirect location
|
||||
response = httpx.Response(301, headers=headers, request=request)
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
response.raise_for_excepted_status([200])
|
||||
assert "Redirect response '301 Moved Permanently'" in str(exc_info.value)
|
||||
assert "Redirect location: 'https://other.org'" in str(exc_info.value)
|
||||
|
||||
# Empty expected list - all status codes should raise
|
||||
response = httpx.Response(200, request=request)
|
||||
with pytest.raises(httpx.HTTPStatusError):
|
||||
response.raise_for_excepted_status([])
|
||||
|
||||
# Calling .raise_for_excepted_status without setting a request instance
|
||||
# should raise a runtime error.
|
||||
response = httpx.Response(200)
|
||||
with pytest.raises(RuntimeError) as exc_info:
|
||||
response.raise_for_excepted_status([200])
|
||||
assert "raise_for_excepted_status" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_response_repr():
|
||||
response = httpx.Response(
|
||||
200,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user