Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
# I am terrible at keeping this up-to-date.

## 2.5.0

- Add common `status_code` and `retry_after` accessors to `WebPushException`
for synchronous and asynchronous responses.

## 2.3.0 (2026-02-09)

- Cleanup from @Rotzbua
Expand Down
5 changes: 5 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,11 @@ try:
)
except WebPushException as ex:
print("I'm sorry, Dave, but I can't do that: {}", repr(ex))
# status_code works with both webpush() and webpush_async().
if ex.status_code in (429, 503):
print("Push service requested a retry after:", ex.retry_after)
# retry_after is either a delay in seconds or an HTTP date:
# https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Retry-After
# Mozilla returns additional information in the body of the response.
if ex.response is not None and ex.response.json():
extra = ex.response.json()
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "pywebpush"
version = "2.4.0"
version = "2.5.0"
# PYTHON_VER
requires-python = ">= 3.10"
license = "MPL-2.0"
Expand Down
26 changes: 25 additions & 1 deletion pywebpush/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,11 @@
class WebPushException(Exception):
"""Web Push failure.

This may contain the requests.Response
This may contain a requests.Response or aiohttp.ClientResponse.

``status_code`` and ``retry_after`` provide a common interface for
inspecting either response type without discarding the original
``response`` object.

"""

Expand All @@ -44,6 +48,26 @@ def __str__(self) -> str:
extra = f", Response {self.response}"
return f"WebPushException: {self.message}{extra}"

@property
def status_code(self) -> int | None:
"""Return the HTTP status for synchronous or asynchronous responses."""
if self.response is None:
return None
return getattr(
self.response,
"status_code",
getattr(self.response, "status", None),
)

@property
def retry_after(self) -> str | None:
"""Return the provider's Retry-After header, when present.

The value can be either a delay in seconds or an HTTP date. See
https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Retry-After
"""
return getattr(self.response, "headers", {}).get("Retry-After", None)


class NoData(Exception):
"""Message contained No Data, no encoding required."""
Expand Down
13 changes: 13 additions & 0 deletions pywebpush/tests/test_webpush.py
Original file line number Diff line number Diff line change
Expand Up @@ -563,6 +563,8 @@ def test_exception(self):

exp = WebPushException("foo")
assert f"{exp}" == "WebPushException: foo"
assert exp.status_code is None
assert exp.retry_after is None
# Really should try to load the response to verify, but this mock
# covers what we need.
response = Mock(spec=Response)
Expand All @@ -577,9 +579,20 @@ def test_exception(self):
response.json.return_value = json.loads(response.text)
response.status_code = 401
response.reason = "Unauthorized"
response.headers = {"Retry-After": "120"}
exp = WebPushException("foo", response)
assert f"{exp}" == "WebPushException: foo, Response {}".format(response.text)
assert f"{exp.response}", "<Response [401]>"
assert cast(requests.Response, exp.response).json().get("errno") == 109
assert exp.status_code == 401
assert exp.retry_after == "120"

async_response = Mock(spec=["status", "headers", "text"])
async_response.status = 503
async_response.headers = {"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}
exp = WebPushException("async failure", async_response)
assert exp.status_code == 503
assert exp.retry_after == "Wed, 21 Oct 2015 07:28:00 GMT"

exp = WebPushException("foo", [1, 2, 3])
assert f"{exp}" == "WebPushException: foo, Response [1, 2, 3]"