# Copyright 2015 Alexey Kinev <rudy@05bit.com>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Base classes for SMS, email, push notifications senders.
"""
import asyncio
[docs]class BaseService:
"""Base sender service class.
Args:
loop: asyncio event loop or Tornado IOLoop
"""
def __init__(self, *, loop):
self.loop = loop
try:
_BaseEventLoop = asyncio.BaseEventLoop
except AttributeError:
_BaseEventLoop = asyncio.AbstractEventLoop
self.is_asyncio = isinstance(loop, _BaseEventLoop)
[docs] def new_http_client(self):
"""Create new http client.
If service is running on Tornado IOLoop, then
`tornado.httpclient.AsyncHTTPClient`_ is used. When asyncio event
loop is specified `aiohttp`_ client is used, in that case aiohttp
library is required to be installed.
.. _aiohttp: http://aiohttp.readthedocs.org
.. _tornado.httpclient.AsyncHTTPClient: http://tornado.readthedocs.org/en/latest/httpclient.html
"""
if self.is_asyncio:
from ._http.aio import AioHTTPClient
return AioHTTPClient(loop=self.loop)
else:
from ._http.tornado import TornadoHTTPClient
return TornadoHTTPClient(loop=self.loop)
[docs]class BasePushService(BaseService):
"""Push notifications sender interface class mixin.
Provides empty methods for push notifications sender, which should
be implemented in subclasses.
Usage example::
from pushka import BasePushService
class ParsePushService(BasePushService):
\"\"\"Send push notifications via Parse service.\"\"\"
@asyncio.coroutine
def send_push(self, token=None, device_type=None, tags=None,
alert=None, badge=None, sound=None, **kwargs):
do_something()
# ...
"""
@asyncio.coroutine
[docs] def add_target(self, token=None, device_type=None, tags=None):
"""Register target device push token. Coroutine."""
raise NotImplementedError
@asyncio.coroutine
[docs] def del_target(self, token=None, device_type=None, tags=None):
"""Unregister target device push token. Coroutine."""
raise NotImplementedError
@asyncio.coroutine
@asyncio.coroutine
@asyncio.coroutine
@asyncio.coroutine
[docs] def send_push(self, token=None, device_type=None, tags=None,
alert=None, badge=None, sound=None, **kwargs):
"""Send push message to single or multiple devices specified by
filtering options. Coroutine.
Args:
token (str): Device token string
device_type (str): Device type: 'ios' / 'android'
tags (list): List of string tags / channels
alert (str): Notification alert text
badge (int, optional): Application badge number to set
sound (str, optional): Sound name to play, use 'default' for system sound
"""
raise NotImplementedError
[docs]class BaseMailService(BaseService):
"""Email sender base class.
Provides empty method for email sender, which should be implemented
in subclasses.
Args:
loop: asyncio event loop or Tornado IOLoop
default_sender (str): default sender's email address
Usage example::
from pushka import BaseMailService
class SESMailService(BaseMailService):
\"\"\"Send email via Amazon SES service.\"\"\"
@asyncio.coroutine
def send_mail(self, subject='', body=None, recipients=None, sender=None,
html=None, attachments=None, reply_to=None,
cc=None, bcc=None, **kwargs):
do_something()
# ...
"""
def __init__(self, *, loop, default_sender=None):
super().__init__(loop=loop)
self.default_sender = default_sender
@asyncio.coroutine
[docs] def send_mail(self, text=None, subject='', recipients=None, sender=None,
html=None, attachments=None, reply_to=None,
cc=None, bcc=None, return_path=None, **kwargs):
"""Send email message. Coroutine.
Args:
text (str): Plain text message in UTF-8 encoding
subject (str): Subject string, must be single line string
recipients (list): List of recipients email adresses
sender (str, optional): Sender address or default will be used
html (str, optional): HTML message in UTF-8 encoding
reply_to (str, optional): Reply-to address
cc (str, optional): Cc list
bcc (str, optional): Bcc list
"""
raise NotImplementedError
[docs]class BaseSMSService(BaseService):
"""SMS sender base class.
Provides empty method for SMS sender, which should be implemented
in subclasses.
Args:
loop: asyncio event loop or Tornado IOLoop
default_sender (str): default sender's phone number
Usage example::
from pushka import BaseSMSService
class TwilioSMSService(BaseSMSService):
\"\"\"Send SMS via Twilio service.\"\"\"
@asyncio.coroutine
def send_sms(self, text=None, recipients=None, sender=None):
...
"""
def __init__(self, *, loop, default_sender=None):
super().__init__(loop=loop)
self.default_sender = default_sender
@asyncio.coroutine
[docs] def send_sms(self, *, text, recipients, sender=None):
"""Send SMS. Coroutine.
Args:
text (str): Plain text message in UTF-8 encoding
recipients (list): List of recipients phone numbers in format ['+13334445577', ...]
sender (str): Sender phone number or default one will be used
"""
raise NotImplementedError