Source code for errbot.core_plugins.wsview
from inspect import getmembers, ismethod
from json import loads
import logging
from flask.app import Flask
from flask.views import View
from flask import request
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 = req.data.decode()
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__
log.info("Checking %s for webhooks", classname)
for name, func in getmembers(obj, ismethod):
if getattr(func, '_err_webhook_uri_rule', False):
log.info("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 = request.data.decode()
response = self.func(data, **kwargs)
return response if response else '' # assume None as an OK response (simplifies the client side)