feat: add asynchronous support to core notifier engines and HTTP providers

README.md

@@ -27,6 +27,10 @@ Via pip:
 ```
 $ pip install notifiers
 ```
+For asynchronous support:
+```
+$ pip install notifiers[async]
+```
 Via homebrew:
 ```
 $ brew install notifiers
@@ -53,6 +57,24 @@ Or:
 <NotificationResponse,provider=Pushover,status=Success>
 ```
 
+# Asynchronous Usage
+
+```python
+>>> import asyncio
+>>> from notifiers import notify_async
+>>> asyncio.run(notify_async('slack', webhook_url='https://...', message='Hello Async!'))
+<Response,provider=Slack,status=Success, errors=None>
+```
+
+Or via a notifier instance:
+```python
+>>> import asyncio
+>>> from notifiers import get_notifier
+>>> p = get_notifier('slack')
+>>> asyncio.run(p.notify_async(webhook_url='https://...', message='Hello Async!'))
+<Response,provider=Slack,status=Success, errors=None>
+```
+
 # From CLI
 
 ```text

notifiers/__init__.py

@@ -1,8 +1,8 @@
 import logging
 
 from ._version import __version__
-from .core import all_providers, get_notifier, notify
+from .core import all_providers, get_notifier, notify, notify_async
 
 logging.getLogger("notifiers").addHandler(logging.NullHandler())
 
-__all__ = ["__version__", "all_providers", "get_notifier", "notify"]
+__all__ = ["__version__", "all_providers", "get_notifier", "notify", "notify_async"]

notifiers/core.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import importlib.machinery
 import importlib.util
 import logging
@@ -278,6 +279,17 @@ def _send_notification(self, data: dict) -> Response:
         :param data: Notification data
         """
 
+    async def _send_notification_async(self, data: dict) -> Response:
+        """
+        The core method to trigger the provider notification asynchronously.
+        By default, runs the synchronous `_send_notification` in the running loop's default executor.
+
+        :param data: Notification data
+        :return: A :class:`~notifiers.core.Response` object
+        """
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self._send_notification, data)
+
     def notify(self, raise_on_errors: bool = False, **kwargs) -> Response:
         """
         The main method to send notifications. Prepares the data via the
@@ -296,6 +308,24 @@ def notify(self, raise_on_errors: bool = False, **kwargs) -> Response:
             rsp.raise_on_errors()
         return rsp
 
+    async def notify_async(self, raise_on_errors: bool = False, **kwargs) -> Response:
+        """
+        The main async method to send notifications. Prepares the data via the
+        :meth:`~notifiers.core.SchemaResource._prepare_data` method and then sends the notification
+        via the async :meth:`~notifiers.core.Provider._send_notification_async` method.
+
+        :param kwargs: Notification data
+        :param raise_on_errors: Should the :meth:`~notifiers.core.Response.raise_on_errors` be invoked immediately
+        :return: A :class:`~notifiers.core.Response` object
+        :raises: :class:`~notifiers.exceptions.NotificationError` if ``raise_on_errors`` is set to True and response
+         contained errors
+        """
+        data = self._process_data(**kwargs)
+        rsp = await self._send_notification_async(data)
+        if raise_on_errors:
+            rsp.raise_on_errors()
+        return rsp
+
 
 class ProviderResource(SchemaResource, ABC):
     """The base class that is used to fetch provider related resources like rooms, channels, users etc."""
@@ -309,10 +339,29 @@ def resource_name(self):
     def _get_resource(self, data: dict):
         pass
 
+    async def _get_resource_async(self, data: dict):
+        """
+        The core method to retrieve the resource asynchronously.
+        By default, runs the synchronous `_get_resource` in the running loop's default executor.
+
+        :param data: Resource query data
+        """
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self._get_resource, data)
+
     def __call__(self, **kwargs):
         data = self._process_data(**kwargs)
         return self._get_resource(data)
 
+    async def get_resource_async(self, **kwargs):
+        """
+        Asynchronously fetches provider related resources like rooms, channels, users etc.
+
+        :param kwargs: Resource query data
+        """
+        data = self._process_data(**kwargs)
+        return await self._get_resource_async(data)
+
     def __repr__(self):
         return f"<ProviderResource,provider={self.name},resource={self.resource_name}>"
 
@@ -428,3 +477,16 @@ def notify(provider_name: str, **kwargs) -> Response:
      will raise notification error
     """
     return get_notifier(provider_name=provider_name, strict=True).notify(**kwargs)
+
+
+async def notify_async(provider_name: str, **kwargs) -> Response:
+    """
+    Quickly sends a notification asynchronously without needing to get a notifier via the :func:`get_notifier` method.
+
+    :param provider_name: Name of the notifier to use. Note that if this notifier name does not exist it will raise a
+    :param kwargs: Notification data, dependant on provider
+    :return: :class:`Response`
+    :raises: :class:`~notifiers.exceptions.NoSuchNotifierError` If ``provider_name`` is unknown,
+     will raise notification error
+    """
+    return await get_notifier(provider_name=provider_name, strict=True).notify_async(**kwargs)

notifiers/providers/dingtalk.py

@@ -106,6 +106,17 @@ def _send_notification(self, data: dict) -> Response:
         params = {"access_token": data["access_token"]}
         payload = self._prepare_data(data)
 
-        response = requests.post(url, params=params, json=payload, headers={"Content-Type": "application/json", "Accept": "application/json"})
+        headers = {"Content-Type": "application/json", "Accept": "application/json"}
+        response, errors = requests.post(url, params=params, json=payload, headers=headers, path_to_errors=self.path_to_errors)
 
-        return self._create_response(response)
+        return self.create_response(payload, response, errors)
+
+    async def _send_notification_async(self, data: dict) -> Response:
+        url = self._prepare_url()
+        params = {"access_token": data["access_token"]}
+        payload = self._prepare_data(data)
+
+        headers = {"Content-Type": "application/json", "Accept": "application/json"}
+        response, errors = await requests.async_post(url, params=params, json=payload, headers=headers, path_to_errors=self.path_to_errors)
+
+        return self.create_response(payload, response, errors)

notifiers/providers/gitter.py

@@ -58,6 +58,27 @@ def _get_resource(self, data: dict) -> list:
         rsp = response.json()
         return rsp["results"] if filter_ else rsp
 
+    async def _get_resource_async(self, data: dict) -> list:
+        headers = self._get_headers(data["token"])
+        filter_ = data.get("filter")
+        params = {"q": filter_} if filter_ else {}
+        response, errors = await requests.async_get(
+            self.base_url,
+            headers=headers,
+            params=params,
+            path_to_errors=self.path_to_errors,
+        )
+        if errors:
+            raise ResourceError(
+                errors=errors,
+                resource=self.resource_name,
+                provider=self.name,
+                data=data,
+                response=response,
+            )
+        rsp = response.json()
+        return rsp["results"] if filter_ else rsp
+
 
 class Gitter(GitterMixin, Provider):
     """Send Gitter notifications"""
@@ -98,3 +119,11 @@ def _send_notification(self, data: dict) -> Response:
         headers = self._get_headers(data.pop("token"))
         response, errors = requests.post(url, json=data, headers=headers, path_to_errors=self.path_to_errors)
         return self.create_response(data, response, errors)
+
+    async def _send_notification_async(self, data: dict) -> Response:
+        room_id = data.pop("room_id")
+        url = self.base_url + self.message_url.format(room_id=room_id)
+
+        headers = self._get_headers(data.pop("token"))
+        response, errors = await requests.async_post(url, json=data, headers=headers, path_to_errors=self.path_to_errors)
+        return self.create_response(data, response, errors)

notifiers/providers/join.py

@@ -1,9 +1,10 @@
 import json
 
-import requests
+import requests as requests_sync
 
 from ..core import Provider, ProviderResource, Response
 from ..exceptions import ResourceError
+from ..utils import requests
 from ..utils.schema.helpers import list_to_commas, one_or_more
 
 
@@ -18,12 +19,12 @@ def _join_request(url: str, data: dict) -> tuple:
         # Can 't use generic requests util since API doesn't always return error status
         errors = None
         try:
-            response = requests.get(url, params=data)
+            response = requests_sync.get(url, params=data)
             response.raise_for_status()
             rsp = response.json()
             if not rsp["success"]:
                 errors = [rsp["errorMessage"]]
-        except requests.RequestException as e:
+        except requests_sync.RequestException as e:
             if e.response is not None:
                 response = e.response
                 try:
@@ -36,6 +37,30 @@ def _join_request(url: str, data: dict) -> tuple:
 
         return response, errors
 
+    @staticmethod
+    async def _join_request_async(url: str, data: dict) -> tuple:
+        errors = None
+        try:
+            response, errors = await requests.async_get(url, params=data, raise_for_status=False)
+            if response is not None:
+                try:
+                    response.raise_for_status()
+                    rsp = response.json()
+                    if not rsp["success"]:
+                        errors = [rsp["errorMessage"]]
+                except Exception as e:
+                    try:
+                        errors = [response.json()["errorMessage"]]
+                    except (json.decoder.JSONDecodeError, ValueError, KeyError):
+                        errors = [response.text or str(e)]
+            else:
+                pass
+        except Exception as e:
+            response = None
+            errors = [str(e)]
+
+        return response, errors
+
 
 class JoinDevices(JoinMixin, ProviderResource):
     """Return a list of Join devices IDs"""
@@ -63,6 +88,19 @@ def _get_resource(self, data: dict):
             )
         return response.json()["records"]
 
+    async def _get_resource_async(self, data: dict):
+        url = self.base_url + self.devices_url
+        response, errors = await self._join_request_async(url, data)
+        if errors:
+            raise ResourceError(
+                errors=errors,
+                resource=self.resource_name,
+                provider=self.name,
+                data=data,
+                response=response,
+            )
+        return response.json()["records"]
+
 
 class Join(JoinMixin, Provider):
     """Send Join notifications"""
@@ -196,3 +234,8 @@ def _send_notification(self, data: dict) -> Response:
         url = self.base_url + self.push_url
         response, errors = self._join_request(url, data)
         return self.create_response(data, response, errors)
+
+    async def _send_notification_async(self, data: dict) -> Response:
+        url = self.base_url + self.push_url
+        response, errors = await self._join_request_async(url, data)
+        return self.create_response(data, response, errors)

notifiers/providers/mailgun.py

@@ -207,3 +207,23 @@ def _send_notification(self, data: dict) -> Response:
             path_to_errors=self.path_to_errors,
         )
         return self.create_response(data, response, errors)
+
+    async def _send_notification_async(self, data: dict) -> Response:
+        base_url = data.pop("base_url")
+        domain = data.pop("domain")
+        url = f"{base_url}/v3/{domain}/messages"
+        auth = "api", data.pop("api_key")
+        files = []
+        if data.get("attachment"):
+            files += requests.file_list_for_request(data["attachment"], "attachment")
+        if data.get("inline"):
+            files += requests.file_list_for_request(data["inline"], "inline")
+
+        response, errors = await requests.async_post(
+            url=url,
+            data=data,
+            auth=auth,
+            files=files,
+            path_to_errors=self.path_to_errors,
+        )
+        return self.create_response(data, response, errors)

notifiers/providers/notify.py

@@ -62,3 +62,19 @@ def _send_notification(self, data: dict) -> Response:
             path_to_errors=self.path_to_errors,
         )
         return self.create_response(data, response, errors)
+
+    async def _send_notification_async(self, data: dict) -> Response:
+        url = self.base_url.format(base_url=data.pop("base_url"))
+        token = data.pop("token", None)
+        headers = self._get_headers(token) if token else {}
+        response, errors = await requests.async_post(
+            url,
+            json={
+                "message": data.pop("message"),
+                "title": data.pop("title", None),
+                "tags": data.pop("tags", []),
+            },
+            headers=headers,
+            path_to_errors=self.path_to_errors,
+        )
+        return self.create_response(data, response, errors)

notifiers/providers/pagerduty.py

@@ -125,3 +125,8 @@ def _send_notification(self, data: dict) -> Response:
         url = self.base_url
         response, errors = requests.post(url, json=data, path_to_errors=self.path_to_errors)
         return self.create_response(data, response, errors)
+
+    async def _send_notification_async(self, data: dict) -> Response:
+        url = self.base_url
+        response, errors = await requests.async_post(url, json=data, path_to_errors=self.path_to_errors)
+        return self.create_response(data, response, errors)

notifiers/providers/popcornnotify.py

@@ -41,3 +41,7 @@ def _prepare_data(self, data: dict) -> dict:
     def _send_notification(self, data: dict) -> Response:
         response, errors = requests.post(url=self.base_url, json=data, path_to_errors=self.path_to_errors)
         return self.create_response(data, response, errors)
+
+    async def _send_notification_async(self, data: dict) -> Response:
+        response, errors = await requests.async_post(url=self.base_url, json=data, path_to_errors=self.path_to_errors)
+        return self.create_response(data, response, errors)

notifiers/providers/pushbullet.py

@@ -39,6 +39,19 @@ def _get_resource(self, data: dict) -> list:
             )
         return response.json()["devices"]
 
+    async def _get_resource_async(self, data: dict) -> list:
+        headers = self._get_headers(data["token"])
+        response, errors = await requests.async_get(self.devices_url, headers=headers, path_to_errors=self.path_to_errors)
+        if errors:
+            raise ResourceError(
+                errors=errors,
+                resource=self.resource_name,
+                provider=self.name,
+                data=data,
+                response=response,
+            )
+        return response.json()["devices"]
+
 
 class Pushbullet(PushbulletMixin, Provider):
     """Send Pushbullet notifications"""
@@ -120,3 +133,13 @@ def _send_notification(self, data: dict) -> Response:
             path_to_errors=self.path_to_errors,
         )
         return self.create_response(data, response, errors)
+
+    async def _send_notification_async(self, data: dict) -> Response:
+        headers = self._get_headers(data.pop("token"))
+        response, errors = await requests.async_post(
+            self.base_url,
+            json=data,
+            headers=headers,
+            path_to_errors=self.path_to_errors,
+        )
+        return self.create_response(data, response, errors)