lib-send-notifications/send_notifications/send_notification.py

"""Client for sending notifications to the internal notification backend.

Usage:
	notifier = NotificationClient()   # resolves the API key once, fails fast if missing
	notifier.send("Title", "Body text", link="https://example.com")
"""

import logging
import os
import time

import requests


from secret_manager import get_secret

__all__ = [
	"NotificationClient",
	"NotificationError",
	"NotificationUnavailableError",
]

logger = logging.getLogger(__name__)

DEFAULT_BACKEND_URL = "http://localhost:8101/internal/receive-notifications"
DEFAULT_KEY_ENV = "NOTIFY_API_KEY"
DEFAULT_KEY_PATH = "/etc/secrets/api_key"
RETRYABLE_STATUS = {429, 502, 503, 504}


class NotificationError(Exception):
	"""The notification could not be delivered."""

	def __init__(self, message: str, status_code: int | None = None):
		super().__init__(message)
		self.status_code = status_code


class NotificationUnavailableError(NotificationError):
	"""The backend was unreachable or kept failing after all retries."""


class NotificationClient:
	"""Sends notifications to the internal backend service.

	The API key is resolved once at construction time and kept in memory, so it
	keeps working after secret files are deleted. Construction raises if no key
	can be found, so misconfiguration surfaces at startup rather than on the
	first notification.

	Args:
		api_key: Internal API key. If omitted, it is read from the environment
			variable ``key_env`` first, then from the file at ``key_path``.
		url: Backend endpoint. Defaults to the BACKEND_API_URL environment
			variable, then to DEFAULT_BACKEND_URL.
		key_env: Name of the environment variable holding the API key.
		key_path: Path of the secret file holding the API key.

	Raises:
		MissingCredentialError: No API key was passed, set in the environment,
			or found in the secret file.
	"""

	def __init__(
		self,
		api_key: str | None = None,
		url: str | None = None,
		*,
		key_env: str = DEFAULT_KEY_ENV,
		key_path: str = DEFAULT_KEY_PATH,
	):
		self._api_key = api_key or get_secret(key_env, key_path)
		self.url = url or os.getenv("BACKEND_API_URL", DEFAULT_BACKEND_URL)

	def __repr__(self) -> str:
		# Never include the API key in logs or tracebacks.
		return f"{type(self).__name__}(url={self.url!r})"

	def send(
		self,
		title: str,
		body: str,
		link: str | None = None,
		category: str = "utility",
		*,
		user_id: int = 1,
		max_attempts: int = 5,
		timeout: float = 5.0,
	) -> str:
		"""Send a notification to the backend.

		Delivery is retried with exponential backoff (1s, 2s, 4s, ...) on
		timeouts, connection errors, and HTTP 429/502/503/504. Other HTTP
		errors fail immediately.

		Args:
			title: Short headline shown for the notification.
			body: Main notification text.
			link: Optional URL the notification should open.
			category: Notification category used by the backend for routing
				and display. Defaults to "utility".
			user_id: ID of the user who receives the notification.
			max_attempts: Total number of delivery attempts, including the
				first.
			timeout: Per-request timeout in seconds.

		Returns:
			The backend's response body as text.

		Raises:
			NotificationUnavailableError: The backend was unreachable or kept
				returning transient errors after all attempts.
			NotificationError: The backend rejected the request (non-retryable
				HTTP error) or the request could not be sent. The HTTP status
				is available as ``status_code`` when there is one.
		"""
		headers = {
			"X-API-Key-Internal": self._api_key,
			"Content-Type": "application/json",
		}
		payload = {
			"user_id": user_id,
			"message": {
				"title": title,
				"body": body,
				"link": link,
				"category": category,
				"timestamp": int(time.time()),
			},
		}

		logger.debug("Sending notification: title=%r link=%r", title, link)
		last_error = "unknown error"

		with requests.Session() as session:
			for attempt in range(1, max_attempts + 1):
				try:
					response = session.post(
						self.url, headers=headers, json=payload, timeout=timeout
					)
				except (requests.Timeout, requests.ConnectionError) as e:
					last_error = type(e).__name__
				except requests.RequestException as e:
					raise NotificationError(f"Request failed: {e}") from e
				else:
					if response.ok:
						logger.info("Notification sent: %r", title)
						return response.text

					logger.warning(
						"Backend returned HTTP %s: %s",
						response.status_code,
						response.text,
					)
					if response.status_code not in RETRYABLE_STATUS:
						raise NotificationError(
							f"Backend rejected notification "
							f"(HTTP {response.status_code})",
							status_code=response.status_code,
						)
					last_error = f"HTTP {response.status_code}"

				logger.warning(
					"Attempt %d/%d failed: %s", attempt, max_attempts, last_error
				)
				if attempt < max_attempts:
					time.sleep(2 ** (attempt - 1))

		raise NotificationUnavailableError(
			f"Notification service unavailable after {max_attempts} attempts: "
			f"{last_error}"
		)