Serving notifications from local and remote data sources
""" | |
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment