Source code for platypush.plugins.http

import logging
import os
import requests

from platypush.message import Message
from platypush.plugins import Plugin, action

logger = logging.getLogger(__name__)


[docs] class HttpPlugin(Plugin): """ Plugin for executing custom HTTP requests. Some example usages:: # Execute a GET request on a JSON endpoint { "type": "request", "action": "http.get", "args": { "url": "http://remote-host/api/v1/entity", "params": { "start": "2000-01-01" } } } # Execute an action on another Platypush host through HTTP interface { "type": "request", "action": "http.post", "args": { "url": "http://remote-host:8008/execute", "json": { "type": "request", "target": "remote-host", "action": "music.mpd.play" } } } """
[docs] def __init__(self, **kwargs): super().__init__(**kwargs)
@staticmethod def _exec(method, url, output='text', **kwargs): """Available output types: text (default), json, binary""" if 'username' in kwargs and 'password' in kwargs: kwargs['auth'] = (kwargs.pop('username'), kwargs.pop('password')) method = getattr(requests, method) response = method(url, **kwargs) response.raise_for_status() if output == 'json': output = response.json() if output == 'binary': output = response.content else: output = response.text try: # If the response is a Platypush JSON, extract it output = Message.build(output) except Exception as e: logger.debug(e) return output
[docs] @action def get(self, url, **kwargs): """ Perform a GET request :param url: Target URL :type url: str :param kwargs: Additional arguments that will be transparently provided to the ``requests`` object, including but not limited to query params, data, JSON, headers etc. (see https://docs.python-requests.org/en/master/user/quickstart/#make-a-request) :type kwargs: dict """ return self._exec(method='get', url=url, **kwargs)
[docs] @action def post(self, url, **kwargs): """ Perform a POST request :param url: Target URL :type url: str :param kwargs: Additional arguments that will be transparently provided to the ``requests`` object, including but not limited to query params, data, JSON, headers etc. (see https://docs.python-requests.org/en/master/user/quickstart/#make-a-request) :type kwargs: dict """ return self._exec(method='post', url=url, **kwargs)
[docs] @action def head(self, url, **kwargs): """ Perform an HTTP HEAD request :param url: Target URL :type url: str :param kwargs: Additional arguments that will be transparently provided to the ``requests`` object, including but not limited to query params, data, JSON, headers etc. (see https://docs.python-requests.org/en/master/user/quickstart/#make-a-request) :type kwargs: dict """ return self._exec(method='head', url=url, **kwargs)
[docs] @action def put(self, url, **kwargs): """ Perform a PUT request :param url: Target URL :type url: str :param kwargs: Additional arguments that will be transparently provided to the ``requests`` object, including but not limited to query params, data, JSON, headers etc. (see https://docs.python-requests.org/en/master/user/quickstart/#make-a-request) :type kwargs: dict """ return self._exec(method='put', url=url, **kwargs)
[docs] @action def delete(self, url, **kwargs): """ Perform a DELETE request :param url: Target URL :type url: str :param kwargs: Additional arguments that will be transparently provided to the ``requests`` object, including but not limited to query params, data, JSON, headers etc. (see https://docs.python-requests.org/en/master/user/quickstart/#make-a-request) :type kwargs: dict """ return self._exec(method='delete', url=url, **kwargs)
[docs] @action def options(self, url, **kwargs): """ Perform an HTTP OPTIONS request :param url: Target URL :type url: str :param kwargs: Additional arguments that will be transparently provided to the ``requests`` object, including but not limited to query params, data, JSON, headers etc. (see https://docs.python-requests.org/en/master/user/quickstart/#make-a-request) :type kwargs: dict """ return self._exec(method='options', url=url, **kwargs)
[docs] @action def download(self, url: str, path: str, **kwargs): """ Locally download the content of a remote URL. :param url: URL to be downloaded. :param path: Path where the content will be downloaded on the local filesystem - must be a file name. """ path = os.path.abspath(os.path.expanduser(path)) content = self._exec(method='get', url=url, output='binary', **kwargs) with open(path, 'wb') as f: f.write(content)
# vim:sw=4:ts=4:et: