Org Web Adapter

Editing hungryroot/local-ntfy-errors.org

:PROPERTIES:
:ID:       94c50995-dd09-49df-a4ef-f43e87ce188f
:END:
#+title: Local Ntfy Errors

This is a snippet that can be added to local settings (in Hungryroot it would be
local_dev_override.py) and you'll get exceptions posted to ntfy

#+begin_src python
import logging
from urllib import request

from environ import environ

###############################################################################################
# These settings need to have been previously defined in the environment-specific settings file
###############################################################################################
ENV: str
LOGGING: dict

env = environ.Env()

class NtfyHandler(logging.Handler):
    """Publish error log records to an ntfy topic."""

def __init__(
        self,
        topic_url: str,
        auth_token: str = "",
        title_prefix: str = "Hungryroot",
        environment: str = "unknown",
        timeout_seconds: float = 3.0,
    ) -> None:
        super().__init__()
        self.topic_url = topic_url
        self.auth_token = auth_token
        self.title_prefix = title_prefix
        self.environment = environment
        self.timeout_seconds = timeout_seconds

def emit(self, record: logging.LogRecord) -> None:
        if not self.topic_url:
            return

try:
            title = f"{self.title_prefix} [{self.environment}] {record.levelname}"[:120]
            if record.exc_info and record.exc_info[1]:
                body = str(record.exc_info[1])
            else:
                body = record.getMessage()
            if len(body) > 4000:
                body = f"{body[:4000]}\n\n(truncated)"

payload = body.encode("utf-8", errors="replace")
            req = request.Request(self.topic_url, data=payload, method="POST")
            req.add_header("Title", title)
            req.add_header("Priority", "default")
            req.add_header("Tags", "rotating_light,warning")
            if self.auth_token:
                req.add_header("Authorization", f"Bearer {self.auth_token}")

with request.urlopen(req, timeout=self.timeout_seconds):
                return
        except Exception:
            # Keep logging non-blocking: never fail app code on notification errors.
            self.handleError(record)

class ErrorLevelFilter(logging.Filter):
    """Allow all ERROR-and-above log records."""

def filter(self, record: logging.LogRecord) -> bool:
        return record.levelno >= logging.ERROR

LOGGING.setdefault("filters", {})["ntfy_error_only"] = {"()": ErrorLevelFilter}
LOGGING["handlers"]["ntfy"] = {
    "level": "ERROR",
    "()": NtfyHandler,
    "filters": ["ntfy_error_only"],
    "topic_url": "https://ntfy.unbl.ink/hroot-notifications",
    "title_prefix": "Hungryroot Exception",
    "environment": ENV,
}
for logger_config in LOGGING.setdefault("loggers", {}).values():
    logger_handlers = logger_config.setdefault("handlers", [])
    if "ntfy" not in logger_handlers:
        logger_handlers.append("ntfy")

root_handlers = LOGGING.setdefault("root", {}).setdefault("handlers", [])
if "ntfy" not in root_handlers:
    root_handlers.append("ntfy")

#+end_src