Skip to content

API

API references

actions

Module defining the Action class as well as it subclasses:

  • ViewAction
  • HttpAction

Action

Superclass for action buttons.

See: ntfy button action documentation

Parameters:

Name Type Description Default
action str

name of the action (e.g. 'view', 'http')

 required
label str

description of the action

 required
url str

where the action redirects

 required
clear bool

if true, the notification is deleted upon click

 False
Source code in ntfy_lite/actions.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class Action:
    """
    Superclass for action buttons.

    See: [ntfy button action documentation](https://ntfy.sh/docs/publish/#action-buttons)

    Args:
      action: name of the action (e.g. 'view', 'http')
      label: description of the action
      url: where the action redirects
      clear: if true, the notification is deleted upon click
    """

    def __init__(self, action: str, label: str, url: str, clear: bool = False):
        validate_url("Action.url", url)

        self.action = action
        self.label = label
        self.url = url
        if clear:
            self.clear = "true"
        else:
            self.clear = "false"

    def _str(self, attrs: typing.Tuple[str, ...]) -> str:
        values = {attr: getattr(self, attr) for attr in attrs}
        return ", ".join(
            [self.action]
            + [f"{attr}={value}" for attr, value in values.items() if value is not None]
        )

HttpAction

Bases: Action

Class encapsulating the information of a view action. See: ntfy http action

Parameters:

Name Type Description Default
label str

arbitrary string

 required
url str

url to which the request should be sent

 required
clear bool

if the ntfy notification should be cleared after the request succeeds

 False
method HttpMethod

GET, POST or PUT

 GET
headers Optional[Mapping[str, str]]

HTTP headers to be passed in the request

 None
body Optional[str]

HTTP body

 None
Source code in ntfy_lite/actions.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class HttpAction(Action):
    """
    Class encapsulating the information of a view action.
    See: [ntfy http action](https://ntfy.sh/docs/publish/#send-http-request)

    Args:
      label: arbitrary string
      url: url to which the request should be sent
      clear: if the ntfy notification should be cleared after the request succeeds
      method: GET, POST or PUT
      headers: HTTP headers to be passed in the request
      body: HTTP body

    """

    def __init__(
        self,
        label: str,
        url: str,
        clear: bool = False,
        method: HttpMethod = HttpMethod.GET,
        headers: typing.Optional[typing.Mapping[str, str]] = None,
        body: typing.Optional[str] = None,
    ):
        super().__init__("http", label, url, clear)
        self.method = method.value
        self.headers = headers
        self.body = body

    def __str__(self) -> str:
        _attrs = ("label", "url", "clear", "method", "body")
        main = self._str(_attrs)
        if not self.headers:
            return main
        headers_str = ", ".join(
            [f"headers.{key}={value}" for key, value in self.headers.items()]
        )
        return main + ", " + headers_str

HttpMethod

Bases: Enum

List of methods supported by instances of HttpAction.

Source code in ntfy_lite/actions.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class HttpMethod(Enum):
    """
    List of methods supported by instances
    of HttpAction.
    """

    GET = auto()
    """ GET http method """

    POST = auto()
    """ POST http method """

    PUT = auto()
    """ PUT http method """

GET = auto() class-attribute instance-attribute

GET http method

POST = auto() class-attribute instance-attribute

POST http method

PUT = auto() class-attribute instance-attribute

PUT http method

ViewAction

Bases: Action

Class encapsulating the information of a view action. See: ntfy view action For arguments: see documentation of the ntfy_lite.actions.Action superclass

Source code in ntfy_lite/actions.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
class ViewAction(Action):
    """
    Class encapsulating the information of a view action.
    See: [ntfy view action](https://ntfy.sh/docs/publish/#open-websiteapp)
    For arguments: see documentation of the [ntfy_lite.actions.Action][] superclass
    """

    def __init__(self, label: str, url: str, clear: bool = False) -> None:
        super().__init__("view", label, url, clear)

    def __str__(self) -> str:
        _attrs = ("label", "url", "clear")
        return self._str(_attrs)

defaults

Module defining level2tags, i.e. a mapping between logging level to emoticons. This mapping is used for the logging handler (ntfy_lite.handler.NtfyHandler)

level2tags: typing.Dict[LoggingLevel, typing.Tuple[str, ...]] = {logging.CRITICAL: ('fire'), logging.ERROR: ('broken_heart'), logging.WARNING: ('warning'), logging.INFO: ('artificial_satellite'), logging.DEBUG: ('speech_balloon'), logging.NOTSET: tuple()} module-attribute

Default mapping from logging level to tags, i.e. tags that will be added to notifications corresponding to the key logging level.

See ntfy_lite.handler.NtfyHandler

demo_logging

ntfy lite: ntfy logging Handler example

demo_push

ntfy lite: notification push examples

error

NtfyError

Bases: Exception

Error thrown when the push of a notification fails.

Attributes status_code: error code returned by the request reason: reason of the failure

Source code in ntfy_lite/error.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
class NtfyError(Exception):
    """
    Error thrown when the push of a notification fails.

    Attributes
      status_code: error code returned by the request
      reason: reason of the failure
    """

    def __init__(self, status_code: int, reason: str):
        self.status_code = status_code
        self.reason = reason

    def __str__(self):
        return f"{self.status_code} ({self.reason})"

handler

Module defining the NtfyHandler class.

The NtfyHandler is a logging handler, i.e. an handler suitable for the python logging package

# Basic usage

import logging
import ntfy_lite as ntfy

ntfy_handler = ntfy.NtfyHandler("my_topic")

logging.basicConfig(
        level=logging.INFO,
        format="[%(levelname)s] %(asctime)s | %(name)s |  %(message)s",
        datefmt="%d-%b-%y %H:%M:%S",
        handlers=(ntfy_handler,),
)

NtfyHandler

Bases: Handler

Subclass of logging.Handler that pushes ntfy notifications.

The notification title will be the record name, and the notification message will be either the record message or a file attachment (depending on the level2filepath argument).

Source code in ntfy_lite/handler.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class NtfyHandler(logging.Handler):
    """Subclass of [logging.Handler](https://docs.python.org/3/library/logging.html#handler-objects)
    that pushes ntfy notifications.

    The notification title will be the record name, and the
    notification message will be either the record message or a
    file attachment (depending on the level2filepath argument).
    """

    def __init__(
        self,
        topic: str,
        url: str = "https://ntfy.sh",
        twice_in_a_row: bool = True,
        error_callback: typing.Optional[
            typing.Callable[[Exception], typing.Any]
        ] = None,
        level2tags: typing.Dict[LoggingLevel, typing.Tuple[str, ...]] = level2tags,
        level2priority: typing.Dict[LoggingLevel, Priority] = level2priority,
        level2filepath: typing.Dict[LoggingLevel, Path] = {},
        level2email: typing.Dict[LoggingLevel, str] = {},
        dry_run: DryRun = DryRun.off,
    ):
        """
        Args:
          topic: Topic on which the notifications will be pushed.
          url: https://ntfy.sh by default.
          twice_in_a_row: If False, if several similar records (similar: same name
            and same message) are emitted, only the first one will result in notification
            being pushed (to avoid the channel to reach the accepted limits of notifications).
          error_callback: It will be called if a NtfyError is raised when pushing a notification.
          level2tags: mapping between logging level and tags to be associated with the notification
          level2priority: mapping between the logging level and the notification priority.
          level2filepath: If for the logging level of the record a corresponding filepath is set,
            the notification will contain no message but a correspondinf file attachment
            (be aware of the size limits, see https://ntfy.sh/docs/publish/#attach-local-file).
          level2email: If an email address is specified for the logging level of the record,
            the ntfy notification will also request a mail to be sent.
          dry_run: For testing. If 'on', no notification will be sent. If 'error', no notification will be sent,
            instead a NtfyError are raised.
        """
        super().__init__()
        self._url = url
        self._topic = topic
        self._last_messages: typing.Optional[typing.Dict[str, str]]
        self._last_messages = None if twice_in_a_row else {}
        self._level2tags = level2tags
        self._level2priority = level2priority
        self._level2filepath = level2filepath
        self._level2email = level2email
        self._error_callback = error_callback
        self._dry_run = dry_run

        for logging_level in level2priority:
            if logging_level not in self._level2priority:
                raise ValueError(
                    f"NtfyHandler, level2priority argument: missing mapping from "
                    f"logging level {logging_level} to ntfy priority level"
                )

    def _is_new_record(self, record: logging.LogRecord) -> bool:
        if self._last_messages is None:
            return True
        try:
            previous_message = self._last_messages[record.name]
        except KeyError:
            self._last_messages[record.name] = record.msg
            return True
        if record.msg == previous_message:
            return False
        self._last_messages[record.name] = record.msg
        return True

    def emit(self, record: logging.LogRecord) -> None:
        """
        Push the record as an ntfy message.
        """
        if self._last_messages and not self._is_new_record(record):
            return
        try:
            filepath = self._level2filepath[record.levelno]
            message = None
        except KeyError:
            filepath = None
            message = record.msg
        try:
            email = self._level2email[record.levelno]
        except KeyError:
            email = None
        try:
            tags = self._level2tags[record.levelno]
        except KeyError:
            tags = tuple()
        try:
            push(
                self._topic,
                record.name,
                message=message,
                priority=self._level2priority[record.levelno],
                tags=tags,
                email=email,
                filepath=filepath,
                url=self._url,
                dry_run=self._dry_run,
            )
        except Exception as e:
            if self._error_callback is not None:
                self._error_callback(e)
            self.handleError(record)

__init__(topic, url='https://ntfy.sh', twice_in_a_row=True, error_callback=None, level2tags=level2tags, level2priority=level2priority, level2filepath={}, level2email={}, dry_run=DryRun.off)

Parameters:

Name Type Description Default
topic str

Topic on which the notifications will be pushed.

 required
url str

https://ntfy.sh by default.

 'https://ntfy.sh'
twice_in_a_row bool

If False, if several similar records (similar: same name and same message) are emitted, only the first one will result in notification being pushed (to avoid the channel to reach the accepted limits of notifications).

 True
error_callback Optional[Callable[[Exception], Any]]

It will be called if a NtfyError is raised when pushing a notification.

 None
level2tags Dict[LoggingLevel, Tuple[str, ...]]

mapping between logging level and tags to be associated with the notification

 level2tags
level2priority Dict[LoggingLevel, Priority]

mapping between the logging level and the notification priority.

 level2priority
level2filepath Dict[LoggingLevel, Path]

If for the logging level of the record a corresponding filepath is set, the notification will contain no message but a correspondinf file attachment (be aware of the size limits, see https://ntfy.sh/docs/publish/#attach-local-file).

 {}
level2email Dict[LoggingLevel, str]

If an email address is specified for the logging level of the record, the ntfy notification will also request a mail to be sent.

 {}
dry_run DryRun

For testing. If 'on', no notification will be sent. If 'error', no notification will be sent, instead a NtfyError are raised.

 off
Source code in ntfy_lite/handler.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    topic: str,
    url: str = "https://ntfy.sh",
    twice_in_a_row: bool = True,
    error_callback: typing.Optional[
        typing.Callable[[Exception], typing.Any]
    ] = None,
    level2tags: typing.Dict[LoggingLevel, typing.Tuple[str, ...]] = level2tags,
    level2priority: typing.Dict[LoggingLevel, Priority] = level2priority,
    level2filepath: typing.Dict[LoggingLevel, Path] = {},
    level2email: typing.Dict[LoggingLevel, str] = {},
    dry_run: DryRun = DryRun.off,
):
    """
    Args:
      topic: Topic on which the notifications will be pushed.
      url: https://ntfy.sh by default.
      twice_in_a_row: If False, if several similar records (similar: same name
        and same message) are emitted, only the first one will result in notification
        being pushed (to avoid the channel to reach the accepted limits of notifications).
      error_callback: It will be called if a NtfyError is raised when pushing a notification.
      level2tags: mapping between logging level and tags to be associated with the notification
      level2priority: mapping between the logging level and the notification priority.
      level2filepath: If for the logging level of the record a corresponding filepath is set,
        the notification will contain no message but a correspondinf file attachment
        (be aware of the size limits, see https://ntfy.sh/docs/publish/#attach-local-file).
      level2email: If an email address is specified for the logging level of the record,
        the ntfy notification will also request a mail to be sent.
      dry_run: For testing. If 'on', no notification will be sent. If 'error', no notification will be sent,
        instead a NtfyError are raised.
    """
    super().__init__()
    self._url = url
    self._topic = topic
    self._last_messages: typing.Optional[typing.Dict[str, str]]
    self._last_messages = None if twice_in_a_row else {}
    self._level2tags = level2tags
    self._level2priority = level2priority
    self._level2filepath = level2filepath
    self._level2email = level2email
    self._error_callback = error_callback
    self._dry_run = dry_run

    for logging_level in level2priority:
        if logging_level not in self._level2priority:
            raise ValueError(
                f"NtfyHandler, level2priority argument: missing mapping from "
                f"logging level {logging_level} to ntfy priority level"
            )

emit(record)

Push the record as an ntfy message.

Source code in ntfy_lite/handler.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def emit(self, record: logging.LogRecord) -> None:
    """
    Push the record as an ntfy message.
    """
    if self._last_messages and not self._is_new_record(record):
        return
    try:
        filepath = self._level2filepath[record.levelno]
        message = None
    except KeyError:
        filepath = None
        message = record.msg
    try:
        email = self._level2email[record.levelno]
    except KeyError:
        email = None
    try:
        tags = self._level2tags[record.levelno]
    except KeyError:
        tags = tuple()
    try:
        push(
            self._topic,
            record.name,
            message=message,
            priority=self._level2priority[record.levelno],
            tags=tags,
            email=email,
            filepath=filepath,
            url=self._url,
            dry_run=self._dry_run,
        )
    except Exception as e:
        if self._error_callback is not None:
            self._error_callback(e)
        self.handleError(record)

ntfy

Module defining the push method, which send a message or the content of a file to an NTFY channel.

DryRun

Bases: Enum

An optional value of DryRun may be passed as an argument to the ntfy_lite.ntfy.push function.

This is meant for testing.

Source code in ntfy_lite/ntfy.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class DryRun(Enum):
    """
    An optional value of DryRun may be passed as an argument to the [ntfy_lite.ntfy.push][] function.

    - If 'off' is passed (default), then the [ntfy_lite.ntfy.push][] function will publish to ntfy.

    - If 'on' is passed, then the [ntfy_lite.ntfy.push][] function will *not* publish to ntfy.

    - If 'error' is passed, then the [ntfy_lite.ntfy.push][] function will raise an [ntfy_lite.error.NtfyError][].

    This is meant for testing.
    """

    on = auto()
    off = auto()
    error = auto()

push(topic, title, message=None, priority=Priority.DEFAULT, tags=[], click=None, email=None, filepath=None, attach=None, icon=None, actions=[], at=None, url='https://ntfy.sh', dry_run=DryRun.off)

Pushes a notification.

# basic usage
import ntfy_lite as ntfy

ntfy.push(
    "my topic", priority=ntfy.Priority.DEFAULT, message="my message"
)

For more documentation of all arguments, visit: https://ntfy.sh/docs/publish/

Parameters:

Name Type Description Default
topic str

the ntfy topic on which to publish

 required
title str

the title of the notification

 required
message Optional[str]

the message. It is optional and if None, then a filepath argument must be provided instead.

 None
priority Priority

the priority of the notification

 DEFAULT
tags emojis

either a string (a single tag) or a list of string (several tags). see supported emojis

 []
click Optional[str]

URL link to be included in the notification

 None
email Optional[str]

address to which the notification should also be sent

 None
filepath Optional[Path]

path to the file to be sent as attachement. It is optional and if None, then a message argument must be provided instead.

 None
icon Optional[str]

URL to an icon to include in the notification

 None
actions Union[Action, Sequence[Action]]

An action is either a ntfy_lite.actions.ViewAction (i.e. a link to a website) or a ntfy_lite.actions.HttpAction (i.e. sending of a HTTP GET, POST or PUT request to a website)

 []
at Optional[str]

to be used for delayed notification, see scheduled delivery

 None
url Optional[str]

ntfy server

 'https://ntfy.sh'
dry_run DryRun

for testing purposes, see ntfy_lite.ntfy.DryRun

 off
Source code in ntfy_lite/ntfy.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def push(
    topic: str,
    title: str,
    message: typing.Optional[str] = None,
    priority: Priority = Priority.DEFAULT,
    tags: typing.Union[str, typing.Iterable[str]] = [],
    click: typing.Optional[str] = None,
    email: typing.Optional[str] = None,
    filepath: typing.Optional[Path] = None,
    attach: typing.Optional[str] = None,
    icon: typing.Optional[str] = None,
    actions: typing.Union[Action, typing.Sequence[Action]] = [],
    at: typing.Optional[str] = None,
    url: typing.Optional[str] = "https://ntfy.sh",
    dry_run: DryRun = DryRun.off,
) -> None:
    """
    Pushes a notification.

    ```python
    # basic usage
    import ntfy_lite as ntfy

    ntfy.push(
        "my topic", priority=ntfy.Priority.DEFAULT, message="my message"
    )
    ```

    For more documentation of all arguments, visit:
    [https://ntfy.sh/docs/publish/](https://ntfy.sh/docs/publish/)

    Args:
      topic: the ntfy topic on which to publish
      title: the title of the notification
      message: the message. It is optional and if None, then a filepath argument must be provided instead.
      priority: the priority of the notification
      tags (i.e. emojis): either a string (a single tag) or a list of string (several tags). see [supported emojis](https://docs.ntfy.sh)
      click: URL link to be included in the notification
      email: address to which the notification should also be sent
      filepath: path to the file to be sent as attachement.
        It is optional and if None, then a message argument must be provided instead.
      icon: URL to an icon to include in the notification
      actions: An action is either a [ntfy_lite.actions.ViewAction][]
        (i.e. a link to a website) or a [ntfy_lite.actions.HttpAction][]
        (i.e. sending of a HTTP GET, POST or PUT request to a website)
      at: to be used for delayed notification, see [scheduled delivery](https://ntfy.sh/docs/publish/#scheduled-delivery)
      url: ntfy server
      dry_run: for testing purposes, see [ntfy_lite.ntfy.DryRun][]
    """

    # the message manager:
    # - checks that either message or filepath is not None
    # - if filepath is not None, data is a file to the path
    # - else data is the UTF-8 conversion of message
    # This context manager makes sure that data get closed
    # (if a file)
    with _DataManager(message, filepath) as data:
        # checking that arguments that are expected to be
        # urls are urls
        urls = {"click": click, "attach": attach, "icon": icon}
        for attr, value in urls.items():
            # throw value error if not None
            # and not a url
            validate_url(attr, value)

        # some argument can be directly set in the
        # headers dict
        direct_mapping: typing.Dict[str, typing.Any] = {
            "Title": title,
            "At": at,
            "Click": click,
            "Email": email,
            "Icon": icon,
        }
        headers = {key: value for key, value in direct_mapping.items() if value}

        # adding priority
        headers["Priority"] = priority.value

        # adding tags
        if tags:
            if isinstance(tags, str):
                tags = (tags,)
            headers["Tags"] = ",".join([str(t) for t in tags])

        # adding actions
        if actions:
            if isinstance(actions, Action):
                actions = [actions]
            headers["Actions"] = "; ".join([str(action) for action in actions])

        # sending
        if dry_run == DryRun.off:
            response = requests.put(f"{url}/{topic}", data=data, headers=headers)
            if not response.ok:
                raise NtfyError(response.status_code, response.reason)
        elif dry_run == DryRun.error:
            raise NtfyError(-1, "DryRun.error passed as argument")

ntfy2logging

Module defining:

  • LoggingLevel: typing union of all logging levels
  • Priority: enumeration over ntfy priority levels
  • level2priority: default mapping between logging levels and ntfy priority levels

LoggingLevel = typing.Literal[logging.DEBUG, logging.INFO, logging.NOTSET, logging.WARNING, logging.ERROR, logging.CRITICAL] module-attribute

Union of all logging levels (DEBUG, INFO, NOTSET, WARNING, ERROR and CRITICAL)

level2priority: typing.Dict[LoggingLevel, Priority] = {logging.CRITICAL: Priority.MAX, logging.ERROR: Priority.HIGH, logging.WARNING: Priority.HIGH, logging.INFO: Priority.DEFAULT, logging.DEBUG: Priority.LOW, logging.NOTSET: Priority.MIN} module-attribute

Default mapping from logging level to ntfy priority level (e.g. a record of level INFO maps to a notification of piority level 3)

Priority

Bases: Enum

Enumeration of supported ntfy priority levels

Source code in ntfy_lite/ntfy2logging.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class Priority(Enum):
    """
    Enumeration of supported ntfy priority levels
    """

    MAX = "5"
    """MAX"""

    HIGH = "4"
    """HIGH"""

    DEFAULT = "3"
    """DEFAULT"""

    LOW = "2"
    """LOW"""

    MIN = "1"
    """MIN"""

DEFAULT = '3' class-attribute instance-attribute

DEFAULT

HIGH = '4' class-attribute instance-attribute

HIGH

LOW = '2' class-attribute instance-attribute

LOW

MAX = '5' class-attribute instance-attribute

MAX

MIN = '1' class-attribute instance-attribute

MIN

utils

Module defining the function 'validate_url'.

validate_url(attribute, value)

Return None if value is a valid URL or is None, raises a ValueError otherwise.

Parameters:

Name Type Description Default
attribute str

an arbitrary string, used in the message of the raised ValueError

 required
value Optional[str]

the string to check

 required
Source code in ntfy_lite/utils.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def validate_url(attribute: str, value: typing.Optional[str]) -> None:
    """
    Return None if value is a valid URL or is None,
    raises a ValueError otherwise.

    Args:
      attribute: an arbitrary string, used in the message of the
        raised ValueError
      value: the string to check
    """
    if value is None:
        return
    if validators.url(value) is not True:
        raise ValueError(f"the value for {attribute} ({value}) is not an url")
    return