Usage

Basic Usage

The easiest way to initialize a notifier is via the get_notifier() helper:

>>> import notifiers
>>> pushover = notifiers.get_notifier('pushover')

Or import it directly:

>>> from notifiers.providers.pushover import Pushover
>>> pushover = Pushover()

To send a notification invoke the notify() method:

>>> pushover.notify(apikey='FOO', user='BAR', message='BAZ')

The notifiers.core.Provider.notify() method takes key word arguments based on the provider’s schema. The message key word is used in all notifiers.

Note

You can also send a notification without getting a provider object via the notifiers.core.notify() method:

>>> from notifiers import notify
>>> notify('pushover', apikey='FOO', user='BAR', message='BAZ').

The first argument of the notify() method is the requested provider name. If such does not exist a NoSuchNotifierError exception will be raised.

If there’s a problem with sent key words, a NotifierException will be thrown:

>>> import notifiers
>>> pushover = notifiers.get_notifier('pushover')
>>> pushover.notify(message='FOO')
Traceback (most recent call last):
  ...
notifiers.exceptions.BadArguments: Error with sent data: 'user' is a required property

In this case, a BadArguments exception was thrown since not all required key words were sent.

Provider schema

Each provider’s arguments are declared as a pydantic model, which validates the data sent to the provider. The model is available via the schema_model attribute:

>>> pushover.schema_model
<class 'notifiers.providers.pushover.PushoverSchema'>

The schema property returns the model’s JSON Schema, as generated by pydantic’s model_json_schema(by_alias=True):

>>> pushover.schema
{'additionalProperties': False, 'properties': {'user': {'description': 'the user/group key (not e-mail address) of your user (or you)', 'oneOf': [{'items': {'type': 'string'}, 'minItems': 1, 'type': 'array'}, {'type': 'string'}], 'title': 'User'}, 'message': {'description': 'your message', 'title': 'Message', 'type': 'string'}, ...}, 'required': ['user', 'message', 'token'], 'title': 'PushoverSchema', 'type': 'object'}

To see the required arguments use the required property:

>>> pushover.required
{'required': ['user', 'message', 'token']}

Some providers have conditional requirements as well, which are enforced when sending. For example Zulip requires message, email, api_key and to, and exactly one of domain or server:

>>> zulip = notifiers.get_notifier('zulip')
>>> zulip.required
{'required': ['message', 'email', 'api_key', 'to']}
>>> zulip.notify(message='hi', email='me@foo.com', api_key='KEY', to='general')
Traceback (most recent call last):
  ...
notifiers.exceptions.BadArguments: Error with sent data: One of 'domain', 'server' is required

To get all of the schema properties, which correlate to the key words it can handle, use arguments:

>>> pushover.arguments['priority']
{'anyOf': [{'maximum': 2, 'minimum': -2, 'type': 'integer'}, {'type': 'null'}], 'default': None, 'description': 'notification priority', 'title': 'Priority'}

Default values are available via defaults:

>>> notifiers.get_notifier('gmail').defaults
{'subject': "New email from 'notifiers'!", 'from': 'notifiers@my-host', 'host': 'smtp.gmail.com', 'port': 587, 'tls': True, 'ssl': False, 'html': False, 'login': True}

Values are converted to the declared type when possible, so priority='1' is accepted and sent as 1.

Environment variables

You can set environment variable to replace any argument that the notifier can use. The default syntax to follow is NOTIFIERS_[PROVIDER_NAME]_[ARGUMENT_NAME]:

$ export NOTIFIERS_PUSHOVER_TOKEN=FOO
$ export NOTIFIERS_PUSHOVER_USER=BAR

Then you could just use:

>>> p.notify(message='message')

Note that you can also set MESSAGE in an environment variable. Environment variable values are converted to the argument’s type, so NOTIFIERS_EMAIL_PORT=587 sets port to 587. Arguments with an alternative spelling, like from_, can be set via either NOTIFIERS_EMAIL_FROM or NOTIFIERS_EMAIL_FROM_. You can also change the default prefix of NOTIFIERS_ by pass the env_prefix argument on notify:

>>> p.notify(message='test', env_prefix='MY_OWN_PREFIX_')

Writing Your Own Providers

The notifiers package is designed to be extensible, allowing you to create and integrate your own notification providers. This guide will walk you through the process of creating a custom provider and making it available to others.

Creating a Custom Provider

To create a custom provider, you need to:

1. Create a new class that inherits from Provider

2. Implement the required methods and schema

3. Register your provider

Here’s a basic example:

from typing import Literal

from pydantic import Field

from notifiers.core import Provider, Response
from notifiers.models import SchemaModel


class MyCustomProviderSchema(SchemaModel):
    message: str = Field(description="the message to send")
    api_key: str = Field(description="your API key")
    msgtype: Literal["text", "html"] = Field("text", description="message type")


class MyCustomProvider(Provider):
    name = "my_provider"
    base_url = "https://api.my-provider.com"
    site_url = "https://my-provider.com"

    # The pydantic model that declares and validates the provider arguments
    schema_model = MyCustomProviderSchema

    def _send_notification(self, data: dict) -> Response:
        # `data` is a dict of the validated arguments, with defaults applied
        # Implement your notification logic here
        return self.create_response(data)

Making Your Provider Installable

To make your provider available to others, you need to:

1. Create a proper Python package structure

2. Define an entry point in your setup.py

Here’s the recommended package structure:

myproject/
├── myproject/
│   ├── __init__.py
│   └── provider.py    # Contains your provider implementation
├── setup.py
├── README.md
└── requirements.txt

Configure your setup.py to register the provider:

from setuptools import setup, find_packages

setup(
    name="myproject",
    version="0.1.0",
    packages=find_packages(),
    install_requires=[
        "notifiers>=2.0.0"
    ],
    # Register your provider as an entry point
    entry_points={
        "notifiers": [
            "my_provider = myproject.provider:MyCustomProvider"
        ]
    },
    author="Your Name",
    author_email="your.email@example.com",
    description="A custom notification provider for notifiers",
    long_description=open("README.md").read(),
    long_description_content_type="text/markdown",
    url="https://github.com/yourusername/myproject",
    classifiers=[
        "Programming Language :: Python :: 3",
        "License :: OSI Approved :: MIT License",
        "Operating System :: OS Independent",
    ],
    python_requires=">=3.10",
)

Using Your Custom Provider

Once installed, your provider can be used like any built-in provider:

>>> from notifiers import get_notifier
>>> notifier = get_notifier('my_provider')
>>> notifier.notify(
...     message='Hello from my custom provider!',
...     api_key='your-api-key'
... )

Testing Your Provider

It’s crucial to thoroughly test your provider. Here’s how to set up tests:

import pytest
from notifiers.core import Provider
from notifiers.exceptions import BadArguments

def test_provider_arguments():
    provider = MyCustomProvider()
    data = {
        "message": "test message",
        "api_key": "test_key"
    }
    rsp = provider.notify(**data)
    assert rsp.status == "Success"

    # Test invalid arguments
    with pytest.raises(BadArguments):
        provider.notify(message="test")  # Missing api_key

Schema Validation

The schema model validates the arguments before _send_notification is called. Invalid arguments raise BadArguments. Here’s a more detailed example:

from pydantic import Field, model_validator
from pydantic_core import PydanticCustomError

from notifiers.core import Provider
from notifiers.models import OneOrMore, SchemaModel, Url


class MyCustomProviderSchema(SchemaModel):
    message: str | None = Field(None, description="The notification message")
    html: str | None = Field(None, description="The notification message, as HTML")
    api_key: str = Field(description="Your API authentication key")
    recipients: OneOrMore[str] = Field(description="One or more recipients")
    callback_url: Url | None = Field(None, alias="callbackUrl", description="Callback URL")
    username: str | None = Field(None, description="Username")
    password: str | None = Field(None, description="Password")
    timeout: int = Field(10, ge=1, description="Request timeout in seconds")

    @model_validator(mode="after")
    def _check_arguments(self):
        # If username is provided, password is required (and vice versa)
        self.require_dependencies({"username": ["password"], "password": ["username"]})
        # At least one of these must be passed
        if not (self.is_set("message") or self.is_set("html")):
            raise PydanticCustomError("message_or_html", "Need either 'message' or 'html'")
        return self

Fields without a default are required. OneOrMore[str] accepts a single string or a non empty list of strings. Field names are snake_case. When the remote API uses a different name, set it as the alias (callbackUrl above): both names are accepted, and the alias is used in the data passed to _send_notification. See notifiers.models for the available field types, and Migrating to 2.0 for how JSON schema concepts map to pydantic.

Troubleshooting Guide

Common issues and solutions when developing providers:

1. Entry Point Not Found

   • Ensure your setup.py entry point exactly matches your provider class

   • Verify the package is installed in development mode (pip install -e .)

   • Check that the provider module is importable

2. Schema Validation Errors

   • Use the notify method’s raise_on_errors parameter to debug

   • Check all required fields are provided

   • Verify data types match the schema

3. Response Handling

   • Always return a Response object from _send_notification, e.g. via self.create_response()

   • Include relevant error messages in the response

   • Handle API rate limits and timeouts

Real-World Example

Here’s a complete example of a provider that sends notifications via a REST API:

from typing import Literal

from pydantic import Field

from notifiers.core import Provider, Response
from notifiers.models import SchemaModel
from notifiers.utils import requests


class RestAPIProviderSchema(SchemaModel):
    message: str = Field(description="the message to send")
    api_key: str = Field(description="your API key")
    endpoint: str = Field(description="the API endpoint to call")
    method: Literal["GET", "POST"] = Field("POST", description="HTTP method")
    timeout: int = Field(10, ge=1, description="request timeout in seconds")


class RestAPIProvider(Provider):
    name = "rest_api"
    base_url = "https://api.example.com"
    site_url = "https://example.com"

    schema_model = RestAPIProviderSchema

    def _send_notification(self, data: dict) -> Response:
        headers = {"Authorization": f"Bearer {data['api_key']}"}
        response, errors = requests.RequestsHelper.request(
            url=f"{self.base_url}/{data['endpoint']}",
            method=data["method"].lower(),
            headers=headers,
            json={"text": data["message"]},
            timeout=data["timeout"],
        )
        return self.create_response(data, response, errors)

Usage example:

>>> notifier = get_notifier('rest_api')
>>> notifier.notify(
...     message="Hello API",
...     api_key="your-api-key",
...     endpoint="notifications/send"
... )

Community Providers

Here are some community-created providers that you can use or reference when building your own:

• notifiers-wecom-provider: A provider for sending notifications to WeCom (企业微信)

These providers serve as excellent examples of how to implement custom notification providers. Feel free to:

• Use them directly in your projects

• Study their implementation for best practices

• Contribute to their development

• Use them as templates for your own providers

If you’ve created a provider, please let us know so we can add it to this list!

Best Practices

1. Declare every argument in the schema model, with a description

2. Implement proper error handling

3. Include comprehensive documentation

4. Add tests for your provider

5. Follow the notifiers package conventions

For more examples and detailed API documentation, refer to the API Documentation section.

Provider resources

Some provider have helper method to enable fetching relevant resources (like rooms, users etc.) To get a list of provider resources use the notifiers.core.Provider.resources() property:

>>> telegram.resources
['updates']

Resource share almost all of their functionality with the Provider class, as they have a schema as well:

>>> telegram.updates
<ProviderResource,provider=telegram,resource=updates>
>>> telegram.updates.schema
{'additionalProperties': False, 'properties': {'token': {'description': 'Bot token', 'title': 'Token', 'type': 'string'}}, 'required': ['token'], 'title': 'TelegramUpdatesSchema', 'type': 'object'}

To invoke the resource, just call it:

>>> telegram.updates()
Traceback (most recent call last):
  ...
notifiers.exceptions.BadArguments: Error with sent data: 'token' is a required property

Oops, forgot to send token. Let’s try again:

>>> telegram.updates(token='foo')
[{'update_id': REDACTED, 'message': {'message_id': REDACTED, 'from': {'id': REDACTED, 'is_bot': False, 'first_name': 'REDACTED', 'last_name': 'REDACTED', 'username': 'REDACTED', 'language_code': 'en-US'}, 'chat': {'id': REDACTED, 'first_name': 'REDACTED', 'last_name': 'REDACTED', 'username': 'REDACTED', 'type': 'private'}, 'date': 1516178366, 'text': 'Ccc'}}]

As can be expected, each provider resource returns a completely different response that correlates to the underlying API command it wraps. In this example, by invoking the notifiers.providers.telegram.Telegram.updates() method, you get a response that shows you which active chat IDs your telegram bot token can send to.

Handling errors

There are two base types of errors in Notifiers, data (or schema) related errors and notification related errors. The former can present as so:

>>> import notifiers
>>> pushover = notifiers.get_notifier('pushover')
>>> pushover.notify(message='FOO')
Traceback (most recent call last):
  ...
notifiers.exceptions.BadArguments: Error with sent data: 'user' is a required property

Here we see that an BadArguments exception was raised instantly, since not all required values were sent. Another example:

>>> pushover.notify(message='FOO', token='TOKEN', user='USER', attachment='/foo')
Traceback (most recent call last):
  ...
notifiers.exceptions.BadArguments: Error with sent data: 'attachment': '/foo' is not a 'valid_file'

Some arguments enforce a specific logic. Here pushover’s attachment argument checks that the file exists.

The exception message describes the first error. All errors are available via the errors attribute:

>>> try:
...     pushover.notify(message='FOO', priority=5)
... except notifiers.exceptions.BadArguments as e:
...     print(e.errors)
[{'loc': ['user'], 'msg': 'Field required', 'type': 'missing'}, {'loc': ['token'], 'msg': 'Field required', 'type': 'missing'}, {'loc': ['priority'], 'msg': 'Input should be less than or equal to 2', 'type': 'less_than_equal'}]

There are also notification based errors:

>>> rsp = pushover.notify(message='FOO', token='BAD TOKEN', user='USER')
>>> rsp.raise_on_errors()
Traceback (most recent call last):
  File "/Users/orcarmi/PycharmProjects/notifiers/poc.py", line 49, in <module>
    raise_on_errors=True)
  File "/Users/orcarmi/PycharmProjects/notifiers/notifiers/core.py", line 276, in notify
    rsp.raise_on_errors()
  File "/Users/orcarmi/PycharmProjects/notifiers/notifiers/core.py", line 48, in raise_on_errors
    raise NotificationError(provider=self.provider, data=self.data, errors=self.errors, response=self.response)
notifiers.exceptions.NotificationError: Notification errors: application token is invalid

Note the default behaviour for Response is not to raise exception on error. You can either use the raise_on_errors() method, or pass raise_on_errors=True to the notification command:

>>> pushover.notify(message='FOO', token='BAD TOKEN', user='USER', raise_on_errors=True)
Traceback (most recent call last):
  File "/Users/orcarmi/PycharmProjects/notifiers/poc.py", line 49, in <module>
    raise_on_errors=True)
  File "/Users/orcarmi/PycharmProjects/notifiers/notifiers/core.py", line 276, in notify
    rsp.raise_on_errors()
  File "/Users/orcarmi/PycharmProjects/notifiers/notifiers/core.py", line 48, in raise_on_errors
    raise NotificationError(provider=self.provider, data=self.data, errors=self.errors, response=self.response)
notifiers.exceptions.NotificationError: Notification errors: application token is invalid

You can also use the ok property:

>>> rsp = pushover.notify(message='FOO', token='BAD TOKEN', user='USER')
>>> rsp.ok
False
>>> rsp.errors
['application token is invalid']