aquifer/net/ntfy.py

import urequests

from . import logging
from .config import config
from .config import secrets


class Ntfy:
    def __init__(self):
        self.hostname = config['hostname']

        host = secrets['ntfy_host']
        topic = secrets['ntfy_topic']
        self.url = f'https://{host}/{topic}'

    def message(self, message):
        result = self.sendMessage(message)

        try:
            if result.status_code != 200:
                self.logError(result)
                return False

            return True
        finally:
            result.close()

    def sendMessage(self, message):
        return urequests.post(self.url, data=f"{self.hostname} - {message}")

    def logError(self, result):
        status = result.status_code
        reason = result.reason.decode('utf-8')

        logging.debug(f'ntfy upload issue: {status} {reason}')