ID

94c50995-dd09-49df-a4ef-f43e87ce188f

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