First version

.gitea/workflows/build-local.yaml

@@ -0,0 +1,25 @@
+name: Build and Publish Package via local runner
+
+on:
+  workflow_dispatch:
+
+jobs:
+  build:
+    runs-on: local
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+      - name: Prepare environment
+        run: |
+          virtualenv -p python3 .venv
+          source .venv/bin/activate
+          pip install build twine
+      - name: Build package
+        run: .venv/bin/python -m build
+      - name: Publish package
+        run: |
+          .venv/bin/python -m twine upload \
+          --repository-url https://git.gansejunge.com/api/packages/notifier/pypi \
+          -u ${{ secrets.PACKAGE_UPLOAD_USERNAME }} \
+          -p ${{ secrets.PACKAGE_UPLOAD_TOKEN }} \
+          dist/*

.gitea/workflows/build.yaml

@@ -0,0 +1,25 @@
+name: Build and Publish Package
+
+on:
+  workflow_dispatch:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+      - name: Prepare environment
+        run: |
+          virtualenv -p python3 .venv
+          source .venv/bin/activate
+          pip install build twine
+      - name: Build package
+        run: .venv/bin/python -m build
+      - name: Publish package
+        run: |
+          .venv/bin/python -m twine upload \
+          --repository-url https://git.gansejunge.com/api/packages/notifier/pypi \
+          -u ${{ secrets.PACKAGE_UPLOAD_USERNAME }} \
+          -p ${{ secrets.PACKAGE_UPLOAD_TOKEN }} \
+          dist/*

pyproject.toml

@@ -0,0 +1,39 @@
+[project]
+name = "lib-send-notifications-webhook"
+version = "0.1.0"
+description = "Shared library to send notifications via webhook"
+requires-python = ">=3.10"
+authors = [{ name = "Florian Gänsejunge" }]
+dependencies = [
+	"requests>=2.31",
+	"lib-shared-models",
+]
+
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.setuptools.package-data]
+shared_models = ["py.typed"]
+
+[tool.ruff]
+line-length = 120
+target-version = "py312"
+[tool.ruff.lint]
+select = [
+	"E", "F", # pycodestyle errors, pyflakes (unused imports/variables, undefined names)
+	"I", # import sorting (replaces isort)
+	"B", # bugbear: likely bugs
+	"ASYNC", # async mistakes
+	"RUF006", # un-referenced asyncio.create_task (tasks can be garbage-collected)
+]
+ignore = ["E501"]
+
+[tool.ruff.format]
+indent-style = "tab"
+
+[project.urls]
+Homepage = "https://git.gansejunge.com/notifier/lib-send-notifications-webhook"

src/send_notifications_webhook/__init__.py

@@ -0,0 +1,18 @@
+import logging
+
+from send_notifications_webhook.core import (
+	InvalidNotificationError,
+	NotificationClient,
+	NotificationError,
+	NotificationUnavailableError,
+)
+
+logging.getLogger(__name__).addHandler(logging.NullHandler())
+__all__ = [
+	"InvalidNotificationError",
+	"NotificationClient",
+	"NotificationError",
+	"NotificationUnavailableError",
+]
+
+

src/send_notifications_webhook/core.py

@@ -0,0 +1,197 @@
+import logging
+import time
+
+import requests
+from shared_models import Category, Message, MessageValidationError, ValidationIssue
+
+logger = logging.getLogger(__name__)
+
+
+RETRYABLE_STATUS = {429, 502, 503, 504}
+
+
+class NotificationError(Exception):
+	"""The notification could not be delivered.
+
+	Attributes:
+		status_code: HTTP status returned by the backend, or None if there
+			was no usable response.
+
+	Catch this to handle every failure from the client:
+
+		try:
+			client.send("Title", "Body")
+		except NotificationError as e:
+			print(f"failed: {e}")
+	"""
+
+	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.
+
+	Retrying later may succeed, unlike InvalidNotificationError.
+	"""
+
+
+class InvalidNotificationError(NotificationError, ValueError):
+	"""The notification failed local validation and was never sent.
+
+	Retrying with the same input will always fail; fix the input instead.
+
+	Attributes:
+		issues: Every problem found, as ValidationIssue objects with
+			`field`, `code` and `message`.
+
+	Example:
+		try:
+			client.send("", "Body", category="news")
+		except InvalidNotificationError as e:
+			for issue in e.issues:
+				print(issue.field, issue.code, issue.message)
+	"""
+
+	def __init__(self, message: str, issues: list[ValidationIssue] | None = None):
+		super().__init__(message)
+		self.issues = list(issues or [])
+
+
+class NotificationClient:
+	"""Sends notifications to the notification webhook.
+
+	Args:
+		api_key: Webhook key sent in the Authorization header.
+		url: Full webhook endpoint, e.g.
+			"https://notifier.example.com/webhook/notifications/myscript".
+
+	Raises:
+		ValueError: If `api_key` or `url` is empty.
+
+	Example:
+		client = NotificationClient(api_key="abc123", url="https://.../myscript")
+		client.send("Backup finished", "Took 42s", category="utility")
+	"""
+
+	def __init__(self, api_key: str, url: str):
+		if not api_key:
+			raise ValueError("api_key is required")
+		if not url:
+			raise ValueError("url is required")
+		self._api_key = api_key
+		self.url = 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 | Category = Category.UTILITY,
+		timestamp: int | None = None,
+		*,
+		max_attempts: int = 5,
+		timeout: float = 5.0,
+	) -> str:
+		"""Validate and send one notification, retrying transient failures.
+
+		Args:
+			title: Headline, 1-255 characters after stripping whitespace.
+			body: Message text, 1-255 characters after stripping whitespace.
+			link: Optional URL opened on tap; must start with http:// or
+				https://. A blank string is treated as no link.
+			category: A Category member or its string value
+				("royal-road", "podcasts", "mixtapes", "utility").
+			timestamp: Unix time in seconds; defaults to now.
+			max_attempts: Total tries (first attempt included) for retryable
+				failures: timeouts, connection errors, HTTP 429/502/503/504.
+				Backoff between tries is 1s, 2s, 4s, ...
+			timeout: Seconds to wait for each HTTP request.
+
+		Returns:
+			The backend's response body as text.
+
+		Raises:
+			InvalidNotificationError: Input failed validation; nothing was sent.
+			NotificationUnavailableError: Every attempt failed with a
+				retryable error.
+			NotificationError: The backend rejected the request with a
+				non-retryable status (e.g. 401, 403, 422); see `status_code`.
+			ValueError: `max_attempts` is less than 1.
+
+		Example:
+			client.send(
+				title="New chapter",
+				body="Chapter 12 is out",
+				link="https://example.com/ch12",
+				category="royal-road",
+			)
+		"""
+		if max_attempts < 1:
+			raise ValueError("max_attempts must be at least 1")
+
+		try:
+			message = Message.parse(
+				{
+					"title": title,
+					"body": body,
+					"link": link,
+					"category": category,
+					"timestamp": timestamp if timestamp is not None else int(time.time()),
+				}
+			)
+		except MessageValidationError as e:
+			raise InvalidNotificationError(str(e), issues=e.issues) from e
+
+		headers = {
+			"Authorization": self._api_key,
+			"Content-Type": "application/json",
+		}
+		payload = {"message": message.to_dict()}
+
+		logger.debug("Sending notification: title=%r link=%r", message.title, message.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", message.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}"
+		)

src/send_notifications_webhook/py.typed

@@ -0,0 +1 @@
+