"""
Utility functions for interacting with the Pushover API.
"""
import logging
import salt.utils.http
import salt.utils.json
from salt.exceptions import CommandExecutionError
from salt.exceptions import SaltInvocationError
log = logging.getLogger(__name__)
API_URL = "https://api.pushover.net"
[docs]
class PushoverAPIError(CommandExecutionError):
"""
Generic exception to render Pushover API errors.
"""
status: int
res: dict
raw: str | None
def __init__(self, status: int, res: dict | None = None, raw: str | None = None):
res = res or {}
self.status = status
self.res = res
self.raw = raw
msg = f"Pushover API Error (HTTP {status}): "
if "errors" in res:
msg += "; ".join(res["errors"])
elif self.raw:
msg += self.raw
else:
msg += "(no further description)"
super().__init__(msg)
[docs]
def query(
endpoint,
method="POST",
*,
data=None,
query_params=None,
token=None,
api_version="1",
header_dict=None,
opts=None,
):
"""
.. versionchanged:: 2.0.0
* Parameters have been reordered.
* Uses JSON request bodies by default.
* Errors result in exceptions.
* Returns the decoded response data only.
Query the Pushover API.
endpoint
API endpoint to query (without ``.json`` suffix), e.g. ``messages``.
.. versionchanged:: 2.0.0
Previously, the first parameter was an internally defined identifier.
This accepts all API paths.
method
HTTP method. Defaults to ``POST``.
data
Request body data.
.. versionchanged:: 2.0.0
Automatically dumped to JSON, unless the ``Content-Type`` header
is set explicitly in ``header_dict``.
query_params
URI query parameter dictionary.
token
Pushover API token.
Optional if already specified in ``data`` or ``query_params``,
depending on the method. Overrides them.
api_version
API version. Used for building query URI. Defaults to ``1``.
header_dict
HTTP request headers to add.
opts
Pass through ``__opts__`` to respect Salt HTTP configuration.
"""
query_params = query_params or {}
decode = method != "DELETE"
if token:
if method == "GET":
query_params["token"] = token
else:
data = data or {}
data["token"] = token
header_dict = header_dict or {}
if method != "GET" and "Content-Type" not in header_dict:
header_dict["Content-Type"] = "application/json"
if data:
data = salt.utils.json.dumps(data)
result = salt.utils.http.query(
f"{API_URL}/{api_version}/{endpoint}.json",
method,
params=query_params,
data=data,
header_dict=header_dict,
decode=decode,
decode_type="json",
text=True,
status=True,
opts=opts,
)
if decode and "dict" not in result:
# Salt does not decode the body if the status indicates an error
try:
result["dict"] = salt.utils.json.loads(result["body"])
except ValueError:
result["dict"] = {}
if result["status"] >= 400:
raise PushoverAPIError(result["status"], result.get("dict"), result.get("text"))
if decode:
return result["dict"]
return result["text"]
[docs]
def validate_sound(sound, token, *, context=None, opts=None):
"""
Validate that a specified sound value exists.
sound
Sound to verify
token
Pushover API token
context
Pass through ``__context__`` to allow caching.
.. versionadded:: 2.0.0
opts
Pass through ``__opts__`` to respect Salt HTTP configuration.
.. versionadded:: 2.0.0
"""
context = context or {}
if "pushover_sounds" in context:
sounds = context["pushover_sounds"]
else:
sounds = query("sounds", "GET", token=token, opts=opts)["sounds"]
context["pushover_sounds"] = sounds
return sound in sounds
[docs]
def validate_user(user, token, device=None, *, context=None, opts=None):
"""
Validate that a user/group ID (key) exists and has at least one active device.
If ``device`` is not falsy, additionally validate that the device exists in the account.
user
User or group name to validate. Required.
token
Pushover API token. Required.
device
Optional device for ``user`` to validate.
If unspecified, checks whether any device is available.
context
Pass through ``__context__`` to allow caching.
.. versionadded:: 2.0.0
opts
Pass through ``__opts__`` to respect Salt HTTP configuration.
.. versionadded:: 2.0.0
"""
ckey = (user, token, device)
context = context or {}
if ckey in context:
return True
payload = {"user": user}
if device:
payload["device"] = device
try:
query(
endpoint="users/validate",
data=payload,
token=token,
opts=opts,
)
except PushoverAPIError as err:
if "invalid" in err.res.get("user", ""):
raise CommandExecutionError(f"Pushover: Invalid user '{user}'") from err
if "invalid" in err.res.get("device", ""):
raise CommandExecutionError(
f"Pushover: Invalid device '{device}' for user '{user}'"
) from err
raise
# Only cache successful lookups to avoid false negatives
context[ckey] = True
return True
[docs]
def post_message(
message,
user,
token,
*,
title=None,
device=None,
priority=0,
expire=None,
retry=None,
sound=None,
opts=None,
):
"""
.. versionadded:: 2.0.0
Send a message to a Pushover user or group.
message
Message text to send. Required.
user
User or group of users to send the message to. Must be a user/group ID (key),
not a name or an email address. Required.
token
Pushover API token. Required.
title
Message title. Defaults to ``Message from Salt``.
device
Name of the device to send the message to. Defaults to all devices of the user.
priority
Message priority (integers between ``-2`` and ``2``).
Defaults to ``0``.
.. note::
Emergency priority (``2``) requires ``expire`` and ``retry`` parameters
to be set.
expire
Stop notifying the user after the specified amount of seconds.
The message is still shown after expiry.
retry
Repeat the notification after this amount of seconds. Minimum: ``30``.
sound
`Notification sound <https://pushover.net/api#sounds>`_ to play. Defaults to user default.
opts
Pass through ``__opts__`` to respect Salt HTTP configuration.
"""
if priority not in range(-2, 3):
raise SaltInvocationError(
f"Invalid priority {priority}. Needs to be an integer between -2 and 2 (inclusive)"
)
if priority == 2 and not (expire and retry):
raise SaltInvocationError(
"Emergency messages require `expire` and `retry` parameters to be set"
)
if retry and retry < 30:
raise SaltInvocationError("`retry` needs to be at least 30 (seconds)")
payload = {
"user": user,
"title": title or "Message from Salt",
"priority": priority,
"message": message,
}
if device is not None:
payload["device"] = device
if expire is not None:
payload["expire"] = expire
if retry is not None:
payload["retry"] = retry
if sound:
payload["sound"] = sound
return query(
endpoint="messages",
token=token,
data=payload,
opts=opts,
)