Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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
You can’t perform that action at this time.