Source code for errbot.core_plugins.wsview

import logging
from inspect import getmembers, ismethod
from json import loads

from flask import request
from import Flask
from flask.views import View

import errbot.core_plugins

log = logging.getLogger(__name__)

[docs]def strip_path(): # strip the trailing slashes on incoming requests request.environ["PATH_INFO"] = request.environ["PATH_INFO"].rstrip("/")
[docs]def try_decode_json(req): data = try: return loads(data) except Exception: return None
[docs]def reset_app(): """Zap everything here, useful for unit tests""" errbot.core_plugins.flask_app = Flask(__name__)
[docs]def route(obj): """Check for functions to route in obj and route them.""" flask_app = errbot.core_plugins.flask_app classname = obj.__class__.__name__"Checking %s for webhooks", classname) for name, func in getmembers(obj, ismethod): if getattr(func, "_err_webhook_uri_rule", False):"Webhook routing %s", func.__name__) form_param = func._err_webhook_form_param uri_rule = func._err_webhook_uri_rule verbs = func._err_webhook_methods raw = func._err_webhook_raw callable_view = WebView.as_view( func.__name__ + "_" + "_".join(verbs), func, form_param, raw ) # Change existing rule. for rule in flask_app.url_map._rules: if rule.rule == uri_rule: flask_app.view_functions[rule.endpoint] = callable_view return # Add a new rule flask_app.add_url_rule( uri_rule, view_func=callable_view, methods=verbs, strict_slashes=False )
[docs]class WebView(View):
[docs] def __init__(self, func, form_param, raw): if form_param is not None and raw: raise Exception( "Incompatible parameters: form_param cannot be set if raw is True" ) self.func = func self.raw = raw self.form_param = form_param self.method_filter = ( lambda obj: ismethod(obj) and self.func.__name__ == obj.__name__ )
[docs] def dispatch_request(self, *args, **kwargs): if self.raw: # override and gives the request directly response = self.func(request, **kwargs) elif self.form_param: content = request.form.get(self.form_param) if content is None: raise Exception( "Received a request on a webhook with a form_param defined, " "but that key (%s) is missing from the request.", self.form_param, ) try: content = loads(content) except ValueError: log.debug("The form parameter is not JSON, return it as a string.") response = self.func(content, **kwargs) else: data = try_decode_json(request) if not data: if hasattr(request, "forms"): data = dict(request.forms) # form encoded else: data = response = self.func(data, **kwargs) return ( response if response else "" ) # assume None as an OK response (simplifies the client side)