mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-08 14:12:27 -05:00
9ba2e8b1bf
* Bump flake8-bugbear from 24.12.12 to 25.10.21 Bumps [flake8-bugbear](https://github.com/PyCQA/flake8-bugbear) from 24.12.12 to 25.10.21. - [Release notes](https://github.com/PyCQA/flake8-bugbear/releases) - [Commits](https://github.com/PyCQA/flake8-bugbear/compare/24.12.12...25.10.21) --- updated-dependencies: - dependency-name: flake8-bugbear dependency-version: 25.10.21 dependency-type: direct:development update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> * Fix flake8-bugbear B042 violations in exception classes * Trigger CI re-run * Regenerate poetry.lock with Poetry 2.1.3 for CI compatibility * Regenerated poetry hash again --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Alex Janousek <alex.janousek@gsa.gov>
93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
from typing import List, Union # noqa: UP035 – Python <3.10 compatibility
|
||
|
||
from requests import RequestException, Response
|
||
|
||
REQUEST_ERROR_STATUS_CODE = 503
|
||
REQUEST_ERROR_MESSAGE = "Request failed"
|
||
|
||
TOKEN_ERROR_GUIDANCE = "See our requirements for JSON Web Tokens \
|
||
at https://docs.notifications.service.gov.uk/rest-api.html#authorisation-header"
|
||
TOKEN_ERROR_DEFAULT_ERROR_MESSAGE = "Invalid token: " + TOKEN_ERROR_GUIDANCE
|
||
|
||
|
||
class TokenError(Exception):
|
||
def __init__(self, message=None, token=None):
|
||
self.message = (
|
||
message + ". " + TOKEN_ERROR_GUIDANCE
|
||
if message
|
||
else TOKEN_ERROR_DEFAULT_ERROR_MESSAGE
|
||
)
|
||
self.token = token
|
||
super().__init__(self.message, token)
|
||
|
||
|
||
class TokenExpiredError(TokenError):
|
||
pass
|
||
|
||
|
||
class TokenAlgorithmError(TokenError):
|
||
def __init__(self):
|
||
super().__init__("Invalid token: algorithm used is not HS256")
|
||
|
||
|
||
class TokenDecodeError(TokenError):
|
||
def __init__(self, message=None):
|
||
super().__init__(message or "Invalid token: signature")
|
||
|
||
|
||
class TokenIssuerError(TokenDecodeError):
|
||
def __init__(self):
|
||
super().__init__("Invalid token: iss field not provided")
|
||
|
||
|
||
class TokenIssuedAtError(TokenDecodeError):
|
||
def __init__(self):
|
||
super().__init__("Invalid token: iat field not provided")
|
||
|
||
|
||
class APIError(Exception):
|
||
def __init__(self, response: Response = None, message: str = None):
|
||
self.response = response
|
||
self._message = message
|
||
super().__init__(response, message)
|
||
|
||
def __str__(self):
|
||
return f"{self.status_code} - {self.message}"
|
||
|
||
@property
|
||
def message(
|
||
self,
|
||
) -> Union[str, List[dict]]: # noqa: UP006, UP007 – Python <3.10 compatibility
|
||
try:
|
||
json_resp = self.response.json() # type: ignore
|
||
return json_resp.get("message", json_resp.get("errors"))
|
||
except (TypeError, ValueError, AttributeError, KeyError):
|
||
return self._message or REQUEST_ERROR_MESSAGE
|
||
|
||
@property
|
||
def status_code(self) -> int:
|
||
try:
|
||
return self.response.status_code # type: ignore
|
||
except AttributeError:
|
||
return REQUEST_ERROR_STATUS_CODE
|
||
|
||
|
||
class HTTPError(APIError):
|
||
@staticmethod
|
||
def create(e: RequestException) -> "HTTPError":
|
||
error = HTTPError(e.response)
|
||
if error.status_code == 503:
|
||
error = HTTP503Error(e.response)
|
||
return error
|
||
|
||
|
||
class HTTP503Error(HTTPError):
|
||
"""Specific instance of HTTPError for 503 errors
|
||
|
||
Used for detecting whether failed requests should be retried.
|
||
"""
|
||
|
||
|
||
class InvalidResponse(APIError):
|
||
pass
|