From 20b07fb02f110ff18f3c568480bd430065e2b5a2 Mon Sep 17 00:00:00 2001 From: Fabio Manganiello Date: Thu, 4 Jan 2018 02:45:23 +0100 Subject: [PATCH] Made an HTTP backend, #27 --- platypush/__init__.py | 8 ---- platypush/__main__.py | 12 ++--- platypush/backend/http/__init__.py | 66 +++++++++++++++++++++++++++ platypush/message/request/__init__.py | 42 +++++++++++++---- platypush/pusher/__main__.py | 15 +++--- requirements.txt | 3 ++ setup.py | 7 +-- 7 files changed, 116 insertions(+), 37 deletions(-) create mode 100644 platypush/backend/http/__init__.py diff --git a/platypush/__init__.py b/platypush/__init__.py index 0e220c809..04b7aea43 100644 --- a/platypush/__init__.py +++ b/platypush/__init__.py @@ -75,14 +75,6 @@ class Daemon(object): msg -- platypush.message.Message instance """ if isinstance(msg, Request): - if msg.action.startswith('procedure.'): - logging.info('Executing procedure request: {}'.format(msg)) - proc_name = msg.action.split('.')[-1] - proc_config = Config.get_procedures()[proc_name] - msg = Procedure.build(name=proc_name, requests=proc_config, backend=msg.backend, id=msg.id) - else: - logging.info('Processing request: {}'.format(msg)) - msg.execute(n_tries=self.n_tries) self.processed_requests += 1 if self.requests_to_process \ diff --git a/platypush/__main__.py b/platypush/__main__.py index 211b5fef0..38caff4e5 100644 --- a/platypush/__main__.py +++ b/platypush/__main__.py @@ -1,14 +1,10 @@ import sys -from . import Daemon, __version__ +from platypush import Daemon, __version__ -def main(args=sys.argv[1:]): - print('Starting platypush v.{}'.format(__version__)) - app = Daemon.build_from_cmdline(args) - app.start() - -if __name__ == '__main__': - main() +print('Starting platypush v.{}'.format(__version__)) +app = Daemon.build_from_cmdline(sys.argv[1:]) +app.start() # vim:sw=4:ts=4:et: diff --git a/platypush/backend/http/__init__.py b/platypush/backend/http/__init__.py new file mode 100644 index 000000000..84f41f245 --- /dev/null +++ b/platypush/backend/http/__init__.py @@ -0,0 +1,66 @@ +import logging +import json + +from multiprocessing import Process +from flask import Flask, abort, jsonify, request + +from platypush.message import Message +from platypush.message.request import Request + +from .. import Backend + + +class HttpBackend(Backend): + """ Example interaction with the HTTP backend to make requests: + $ curl -XPOST \ + -d 'msg={"type":"request","target":"volta","action":"tts.say","args": {"phrase":"This is a test"}}' \ + http://localhost:8008 """ + + def __init__(self, port=8008, token=None, **kwargs): + super().__init__(**kwargs) + self.port = port + self.token = token + + + def send_message(self, msg): + raise NotImplementedError('Use cURL or any HTTP client to query the HTTP backend') + + + def _start_server(self): + def app_main(): + app = Flask(__name__) + + @app.route('/', methods=['POST']) + def index(): + args = { k:v for (k,v) in request.form.items() } + + if self.token: + if 'token' not in args or args['token'] != self.token: + abort(401) + + if 'msg' not in args: + abort(400) + + msg = Message.build(args['msg']) + logging.debug('Received message on HTTP backend: {}'.format(msg)) + + if isinstance(msg, Request): + response = msg.execute(async=False) + return str(response) + + return jsonify({ 'status': 'ok' }) + + app.run(debug=True, host='0.0.0.0', port=self.port) + logging.info('Initialized HTTP backend on port {}'.format(self.port)) + + return app_main + + def run(self): + super().run() + self.server_proc = Process(target=self._start_server()) + self.server_proc.start() + self.server_proc.join() + + +# vim:sw=4:ts=4:et: + diff --git a/platypush/message/request/__init__.py b/platypush/message/request/__init__.py index f506bf37f..5bc1a0160 100644 --- a/platypush/message/request/__init__.py +++ b/platypush/message/request/__init__.py @@ -51,15 +51,33 @@ class Request(Message): id += '%.2x' % random.randint(0, 255) return id - def execute(self, n_tries=1): + + def _execute_procedure(self, *args, **kwargs): + from config import Config + + logging.info('Executing procedure request: {}'.format(procedure)) + proc_name = self.action.split('.')[-1] + proc_config = Config.get_procedures()[proc_name] + proc = Procedure.build(name=proc_name, requests=proc_config, backend=self.backend, id=self.id) + proc.execute(*args, **kwargs) + + + def execute(self, n_tries=1, async=True): """ Execute this request and returns a Response object Params: n_tries -- Number of tries in case of failure before raising a RuntimeError + async -- If True, the request will be run asynchronously and the + response posted on the bus when available (default), + otherwise the current thread will wait for the response + to be returned synchronously. """ - def _thread_func(n_tries): - (module_name, method_name) = get_module_and_method_from_action(self.action) + def _thread_func(n_tries): + if self.action.startswith('procedure.'): + return self._execute_procedure(n_tries=n_tries) + + (module_name, method_name) = get_module_and_method_from_action(self.action) plugin = get_plugin(module_name) try: @@ -80,14 +98,20 @@ class Request(Message): _thread_func(n_tries-1) return finally: - # Send the response on the backend - if self.backend and self.origin: - self.backend.send_response(response=response, request=self) + if async: + # Send the response on the backend + if self.backend and self.origin: + self.backend.send_response(response=response, request=self) + else: + logging.info('Response whose request has no ' + + 'origin attached: {}'.format(response)) else: - logging.info('Response whose request has no ' + - 'origin attached: {}'.format(response)) + return response - Thread(target=_thread_func, args=(n_tries,)).start() + if async: + Thread(target=_thread_func, args=(n_tries,)).start() + else: + return _thread_func(n_tries) def __str__(self): diff --git a/platypush/pusher/__main__.py b/platypush/pusher/__main__.py index f80560094..34bdf2b54 100644 --- a/platypush/pusher/__main__.py +++ b/platypush/pusher/__main__.py @@ -2,17 +2,14 @@ import sys from . import Pusher -def main(args=sys.argv[1:]): - opts = Pusher.parse_build_args(args) - pusher = Pusher(config_file=opts.config, backend=opts.backend) +opts = Pusher.parse_build_args(sys.argv[1:]) +pusher = Pusher(config_file=opts.config, backend=opts.backend) - if opts.type == 'event': - pusher.send_event(target=opts.target, type=opts.event, **opts.args) - else: - pusher.send_request(target=opts.target, action=opts.action, timeout=opts.timeout, **opts.args) +if opts.type == 'event': + pusher.send_event(target=opts.target, type=opts.event, **opts.args) +else: + pusher.send_request(target=opts.target, action=opts.action, timeout=opts.timeout, **opts.args) -main() - # vim:sw=4:ts=4:et: diff --git a/requirements.txt b/requirements.txt index b97879695..c65c0d08a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,6 +8,9 @@ kafka-python # PushBullet backend support websocket-client +# HTTP backend support +flask + # Philips Hue plugin support phue diff --git a/setup.py b/setup.py index 5787887e7..e158908e6 100755 --- a/setup.py +++ b/setup.py @@ -43,8 +43,8 @@ setup( packages = find_packages(), entry_points = { 'console_scripts': [ - 'platypush=platypush.__main__:main', - 'pusher=platypush.pusher.__main__:main', + 'platypush=platypush.__main__', + 'pusher=platypush.pusher.__main__', ], }, data_files = [ @@ -63,6 +63,7 @@ setup( extras_require = { 'Support for Apache Kafka backend': ['kafka-python'], 'Support for Pushbullet backend': ['requests', 'websocket-client'], + 'Support for HTTP backend': ['flask'], 'Support for Philips Hue plugin': ['phue'], 'Support for MPD/Mopidy music server plugin': ['python-mpd2'], 'Support for Belkin WeMo Switch plugin': ['ouimeaux'], @@ -70,7 +71,7 @@ setup( 'Support for OMXPlayer plugin': ['omxplayer'], 'Support for YouTube in the OMXPlayer plugin': ['youtube-dl'], 'Support for Google Assistant': ['google-assistant-sdk[samples]'], - 'Support for Flic buttons': ['-e git+https://github.com/50ButtonsEach/fliclib-linux-hci'] + # 'Support for Flic buttons': ['-e git+https://github.com/50ButtonsEach/fliclib-linux-hci'] }, )