Serving notifications from local and remote data sources

new_involved.py

"""
A bit more involved demonstration of __new__ method usage.

For more information see: https://iliazaitsev.me/blog/2018/02/14/python-new
"""
import io
import abc
import json
from os.path import exists
from textwrap import dedent
from threading import Thread
from datetime import datetime
from urllib.parse import parse_qs
from urllib.error import HTTPError
from weakref import WeakValueDictionary
from urllib.request import urlopen, Request
from http.server import HTTPServer, BaseHTTPRequestHandler


_dispatchers = WeakValueDictionary()


class NotificationsDispatcher(metaclass=abc.ABCMeta):
    """
    Class that retrieves a list of notifications for a specific user.

    Usually, notifications are retrieved from the remote server, but for testing
    purposes and local runs it supports reading messages from local source.
    """

    def __new__(cls, user_id: int, method: str= 'http', **kwargs):
        if issubclass(cls, NotificationsDispatcher):
            cls = get_dispatcher(method)
        return object.__new__(cls)

    def __init__(self, user_id: int, dateformat='%m/%d/%Y %H:%M:%S', **kwargs):
        self.user_id = user_id
        self.dateformat = dateformat

    @abc.abstractmethod
    def get_notifications(self) -> dict:
        """
        Returns a list of pending notification messages.
        """

class _LocalDispatcher(NotificationsDispatcher):
    """
    Local notifications dispatcher to retrieve messages from local file.

    Expects a list of notifications in the following CSV format:
    user_id  datetime             message
    1        01/01/2018 09:23:27  Welcome to the app!
    1        01/01/2018 09:23:28  This tutorial should help you get familiar with UI
    2        01/12/2018 14:47:12  Welcome to the app!
    ...
    """

    def __init__(self, user_id: int, filename: str, **kwargs):

        super().__init__(user_id, **kwargs)
        self.filename = filename

    def get_notifications(self):
        notifications = []

        if isinstance(self.filename, str):
            if not exists(self.filename):
                return {'error': 'File does not exist: %s' % self.filename}
            file = open(self.filename)
        else:
            file = self.filename

        # skip header
        _ = next(file)

        for line in file:
            user_id, datestr, message = line.strip().split(',')
            date = datetime.strptime(datestr, self.dateformat)
            if int(user_id) == self.user_id:
                record = {'date': date, 'msg': message}
                notifications.append(record)

        return notifications


class _HTTPDispatcher(NotificationsDispatcher):
    """
    Remote notifications dispatcher retrieving list of notifications from
    HTTP-server.
    """

    def __init__(self, user_id: int, server_url: str, **kwargs):
        super().__init__(user_id, **kwargs)
        self.server_url = server_url

    def get_notifications(self):
        request = Request(self.server_url + '?user_id=%d' % self.user_id)
        request.add_header('Content-Type', 'application/json')
        request.add_header('Accept', 'application/json')

        try:
            response = urlopen(request)
        except HTTPError as e:
            return {"error": e.msg}

        raw_response = json.loads(response.read().decode('utf8'))
        notifications = []
        for entry in raw_response['notifications']:
            date = datetime.strptime(entry['date'], self.dateformat)
            notifications.append({'date': date, 'msg': entry['msg']})

        return notifications


def register_dispatcher(name, cls):
    """
    Public API which is used to register new dispatcher class.
    """
    global _dispatchers
    _dispatchers[name] = cls


def get_dispatcher(name):
    """
    Public API to access dictionary with registered dispatchers.
    """
    if name not in _dispatchers:
        raise ValueError('dispatcher with name \'%s\' is not found' % name)
    return _dispatchers[name]


register_dispatcher('local', _LocalDispatcher)
register_dispatcher('http', _HTTPDispatcher)


# -----------------------------------------------------------------------------
#                           Dummy server implementation
# -----------------------------------------------------------------------------


class NotificationsRequestHandler(BaseHTTPRequestHandler):

    NOTIFICATIONS = {
        1: [
            {
                'msg': 'Welcome to the app!',
                'date': '01/01/2018 10:00:00'
            },
            {
                'msg': 'This tutorial should help you get familiar with UI',
                'date': '01/01/2018 10:00:01'
            }
        ],
        2: [
            {
                'msg': 'Please renew your subscription',
                'date': '01/02/2018 12:30:00'
            }
        ],
        3: []
    }

    def do_GET(self):
        self._set_headers()
        params = parse_qs(self.path[2:])
        notifications = self.NOTIFICATIONS[int(params['user_id'][0])]
        response_obj = {'notifications': notifications}
        response_str = json.dumps(response_obj)
        self.wfile.write(response_str.encode(encoding='utf8'))

    def do_HEAD(self):
        self._set_headers()

    def _set_headers(self):
        self.send_response(200)
        self.send_header('Content-Type', 'text/json')
        self.end_headers()


class ServerThread(Thread):

    def __init__(self, address, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.address = address

    def run(self):
        server = HTTPServer(self.address, NotificationsRequestHandler)
        server.serve_forever()


# -----------------------------------------------------------------------------
#                    Local and remote implementations tests
# -----------------------------------------------------------------------------


def test_remote_dispatcher():
    server_address = ('localhost', 9090)
    server_url = 'http://%s:%s/' % server_address
    server_thread = ServerThread(server_address, daemon=True)
    server_thread.start()
    remote_client = NotificationsDispatcher(
        user_id=1, method='http', server_url=server_url)
    notifications = remote_client.get_notifications()
    print(notifications)


def test_local_dispatcher():
    filename = io.StringIO(dedent("""
        user_id,datetime,message
        1,01/01/2018 09:23:27,Welcome to the app!
        1,01/01/2018 09:23:28,This tutorial should help you get familiar with UI
        2,01/12/2018 14:47:12,Welcome to the app!
    """).strip('\n'))
    local_client = NotificationsDispatcher(
        user_id=2, method='local', filename=filename)
    notifications = local_client.get_notifications()
    print(notifications)


def main():
    test_remote_dispatcher()
    test_local_dispatcher()


if __name__ == '__main__':
    main()