Created
October 4, 2025 16:09
-
-
Save shbatm/5a2e884732c581ff1ffe418de94e85f2 to your computer and use it in GitHub Desktop.
Pushover Python Logger
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """pushover_logger | |
| Lightweight helper to buffer Python logging and send the accumulated | |
| messages to Pushover when the process exits. Designed for ad-hoc scripts | |
| and cron jobs where a concise, monospace summary is useful. | |
| Usage example: | |
| from pushover_logger import setup_logger | |
| log = setup_logger(__name__, pushover_title="Daily job run") | |
| log.info("Starting task") | |
| log.warning("Non-critical issue") | |
| log.error("Example failure") | |
| On process exit the buffered logs are posted to Pushover. Tokens are read | |
| from the environment variables `PUSHOVER_TOKEN` and `PUSHOVER_USER_KEY`. | |
| """ | |
| import atexit | |
| import logging | |
| import os | |
| import textwrap | |
| import urllib.parse | |
| import urllib.request | |
| PUSHOVER_API = "https://api.pushover.net/1/messages.json" | |
| MAX_MSG = 1024 # Pushover message field limit | |
| class PushoverBufferingHandler(logging.Handler): | |
| def __init__( | |
| self, | |
| token: str, | |
| user_key: str, | |
| title: str = None, | |
| device: str = None, | |
| priority: int = 0, | |
| ): | |
| super().__init__() | |
| if not token or not user_key: | |
| raise ValueError("Pushover token and user key are required") | |
| self.token = token | |
| self.user_key = user_key | |
| self.title = title | |
| self.device = device | |
| self.priority = str(priority) | |
| self._buffer = [] | |
| self._flushed = False | |
| def emit(self, record: logging.LogRecord) -> None: | |
| try: | |
| self._buffer.append(self.format(record)) | |
| except Exception: | |
| pass # never raise from logging | |
| def _post(self, data: dict) -> None: | |
| encoded = urllib.parse.urlencode(data).encode("utf-8") | |
| req = urllib.request.Request(PUSHOVER_API, data=encoded) | |
| try: | |
| with urllib.request.urlopen(req, timeout=15): | |
| pass | |
| except Exception: | |
| pass | |
| def flush(self) -> None: | |
| if self._flushed or not self._buffer: | |
| return # skip empty or already sent | |
| self._flushed = True | |
| full = "\n".join(self._buffer) | |
| chunks = textwrap.wrap( | |
| full, width=MAX_MSG, break_long_words=True, replace_whitespace=False | |
| ) | |
| total = len(chunks) | |
| for i, chunk in enumerate(chunks, 1): | |
| data = { | |
| "token": self.token, | |
| "user": self.user_key, | |
| "message": chunk, | |
| "monospace": "1", # default monospace format | |
| } | |
| if self.title: | |
| data["title"] = ( | |
| self.title if total == 1 else f"{self.title} [{i}/{total}]" | |
| ) | |
| if self.device: | |
| data["device"] = self.device | |
| if self.priority: | |
| data["priority"] = self.priority | |
| self._post(data) | |
| def close(self) -> None: | |
| try: | |
| self.flush() | |
| finally: | |
| super().close() | |
| def setup_logger( | |
| name: str = "", | |
| level: int = logging.INFO, | |
| add_console: bool = True, | |
| pushover_title: str = "Script log", | |
| pushover_priority: int = 0, | |
| ) -> logging.Logger: | |
| """ | |
| Reads PUSHOVER_TOKEN and PUSHOVER_USER_KEY from environment. | |
| Sends buffered logs in monospace format on exit. | |
| """ | |
| token = os.getenv("PUSHOVER_TOKEN", "SET_DFAULT_TOKEN") | |
| user = os.getenv("PUSHOVER_USER_KEY", "SET_DEFAULT_USER") | |
| logger = logging.getLogger(name) | |
| logger.setLevel(level) | |
| # fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") | |
| fmt = logging.Formatter("%(levelname)s: %(message)s") | |
| if add_console: | |
| sh = logging.StreamHandler() | |
| sh.setFormatter(fmt) | |
| logger.addHandler(sh) | |
| if token and user: | |
| ph = PushoverBufferingHandler( | |
| token=token, | |
| user_key=user, | |
| title=pushover_title, | |
| priority=pushover_priority, | |
| ) | |
| ph.setFormatter(fmt) | |
| logger.addHandler(ph) | |
| atexit.register(ph.flush) | |
| return logger |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment