Skip to content

Instantly share code, notes, and snippets.

Created February 15, 2018 09:42
Show Gist options
  • Save devforfu/63fa7efe18133dc12f12a2e9ecbe9db4 to your computer and use it in GitHub Desktop.
Save devforfu/63fa7efe18133dc12f12a2e9ecbe9db4 to your computer and use it in GitHub Desktop.
Serving notifications from local and remote data sources
A bit more involved demonstration of __new__ method usage.
For more information see:
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
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)
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}
return notifications
class _HTTPDispatcher(NotificationsDispatcher):
Remote notifications dispatcher retrieving list of notifications from
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')
response = urlopen(request)
except HTTPError as e:
return {"error": e.msg}
raw_response = json.loads('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):
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):
params = parse_qs(self.path[2:])
notifications = self.NOTIFICATIONS[int(params['user_id'][0])]
response_obj = {'notifications': notifications}
response_str = json.dumps(response_obj)
def do_HEAD(self):
def _set_headers(self):
self.send_header('Content-Type', 'text/json')
class ServerThread(Thread):
def __init__(self, address, *args, **kwargs):
super().__init__(*args, **kwargs)
self.address = address
def run(self):
server = HTTPServer(self.address, NotificationsRequestHandler)
# -----------------------------------------------------------------------------
# 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)
remote_client = NotificationsDispatcher(
user_id=1, method='http', server_url=server_url)
notifications = remote_client.get_notifications()
def test_local_dispatcher():
filename = io.StringIO(dedent("""
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!
local_client = NotificationsDispatcher(
user_id=2, method='local', filename=filename)
notifications = local_client.get_notifications()
def main():
if __name__ == '__main__':
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment