import json
import logging
import venusian
from pyramid.exceptions import ConfigurationError
from pyramid.httpexceptions import HTTPForbidden
from pyramid.httpexceptions import HTTPNotFound
from pyramid.renderers import null_renderer
from pyramid.renderers import render
from pyramid.response import Response
from pyramid.security import NO_PERMISSION_REQUIRED
from pyramid_rpc.compat import is_nonstr_iter
from pyramid_rpc.mapper import MapplyViewMapper
from pyramid_rpc.mapper import ViewMapperArgsInvalid
from pyramid_rpc.util import combine
log = logging.getLogger(__name__)
_marker = object()
[docs]class JsonRpcError(Exception):
code = -32603 # sane default
message = 'internal error' # sane default
data = None
def __init__(self, code=None, message=None, data=None):
if code is not None:
self.code = code
if message is not None:
self.message = message
if data is not None:
self.data = data
def as_dict(self):
"""Return a dictionary representation of this object for
serialization in a JSON-RPC response."""
error = dict(code=self.code,
message=self.message)
if self.data is not None:
error['data'] = self.data
return error
[docs]class JsonRpcParseError(JsonRpcError):
code = -32700
message = 'parse error'
[docs]class JsonRpcRequestInvalid(JsonRpcError):
code = -32600
message = 'invalid request'
[docs]class JsonRpcMethodNotFound(JsonRpcError):
code = -32601
message = 'method not found'
[docs]class JsonRpcParamsInvalid(JsonRpcError):
code = -32602
message = 'invalid params'
[docs]class JsonRpcInternalError(JsonRpcError):
code = -32603
message = 'internal error'
def make_error_response(request, error, id=None):
""" Marshal a Python Exception into a ``Response`` object with a
body that is a JSON string suitable for use as a JSON-RPC response
with a content-type of ``application/json`` and return the response.
"""
# we may need to render a parse error, at which point we don't know
# much about the request
renderer = getattr(request, 'rpc_renderer', DEFAULT_RENDERER)
out = {
'jsonrpc': '2.0',
'id': id,
'error': error.as_dict(),
}
body = render(renderer, out, request=request).encode('utf-8')
response = Response(body, charset='utf-8')
response.content_type = 'application/json'
return response
def exception_view(exc, request):
rpc_id = getattr(request, 'rpc_id', None)
if isinstance(exc, JsonRpcError):
fault = exc
log.debug('json-rpc error rpc_id:%s "%s"',
rpc_id, exc.message)
elif isinstance(exc, HTTPNotFound):
fault = JsonRpcMethodNotFound()
log.debug('json-rpc method not found rpc_id:%s "%s"',
rpc_id, request.rpc_method)
elif isinstance(exc, HTTPForbidden):
fault = JsonRpcRequestInvalid()
log.debug('json-rpc method forbidden rpc_id:%s "%s"',
rpc_id, request.rpc_method)
elif isinstance(exc, ViewMapperArgsInvalid):
fault = JsonRpcParamsInvalid()
log.debug('json-rpc invalid method params')
else:
fault = JsonRpcInternalError()
log.exception('json-rpc exception rpc_id:%s "%s"', rpc_id, exc)
return make_error_response(request, fault, rpc_id)
def make_response(request, result):
rpc_id = getattr(request, 'rpc_id', None)
response = request.response
# store content_type before render is called
ct = response.content_type
out = {
'jsonrpc': '2.0',
'id': rpc_id,
'result': result,
} if request.rpc_id is not None else ''
response.body = render(
request.rpc_renderer, out, request=request
).encode(response.charset)
if ct == response.default_content_type:
response.content_type = 'application/json'
return response
def _render(value, system):
return json.dumps(value)
def jsonrpc_renderer(info):
return _render
class jsonrpc_view(object):
""" Decorator that wraps a view and converts the result into a valid
JSON-RPC Response object.
"""
def __init__(self, wrapped):
self.wrapped = wrapped
def __call__(self, context, request):
result = self.wrapped(context, request)
if not request.is_response(result):
result = make_response(request, result)
return result
def parse_request_GET(request):
""" Parse JSON-RPC parameters from the request query string."""
args = request.GET.get('params')
if args is not None:
try:
request.rpc_args = json.loads(args)
except ValueError:
raise JsonRpcParseError
else:
request.rpc_args = ()
request.rpc_method = request.GET.get('method')
request.rpc_id = request.GET.get('id')
request.rpc_version = request.GET.get('jsonrpc')
def parse_request_POST(request):
""" Parse JSON-RPC parameters from the request body."""
try:
body = request.json_body
except ValueError:
raise JsonRpcParseError
request.rpc_id = body.get('id')
request.rpc_args = body.get('params', ())
request.rpc_method = body.get('method')
request.rpc_version = body.get('jsonrpc')
def setup_request(endpoint, request):
""" Parse a JSON-RPC request body."""
if request.method == 'GET':
parse_request_GET(request)
elif request.method == 'POST':
parse_request_POST(request)
else:
log.debug('unsupported request method "%s"', request.method)
raise JsonRpcRequestInvalid
if request.rpc_version != '2.0':
log.debug('id:%s invalid rpc version %s',
request.rpc_id, request.rpc_version)
raise JsonRpcRequestInvalid
if request.rpc_method is None:
log.debug('id:%s invalid rpc method', request.rpc_id)
raise JsonRpcRequestInvalid
log.debug('handling id:%s method:%s',
request.rpc_id, request.rpc_method)
DEFAULT_RENDERER = 'pyramid_rpc:jsonrpc'
class EndpointPredicate(object):
def __call__(self, info, request):
# find the endpoint info
key = info['route'].name
endpoint = request.registry.rpc_endpoints[key]
# potentially setup either rpc v1 or v2 from the parsed body
setup_request(endpoint, request)
# update request with endpoint information
request.rpc_endpoint = endpoint
# Always return True so that even if it isn't a valid RPC it
# will fall through to the notfound_view which will still
# return a valid JSON-RPC response.
return True
class MethodPredicate(object):
def __init__(self, method, renderer):
self.method = method
self.renderer = renderer
def __call__(self, context, request):
if getattr(request, 'rpc_method') == self.method:
request.rpc_renderer = self.renderer
return True
class Endpoint(object):
def __init__(self, name, default_mapper, default_renderer):
self.name = name
self.default_mapper = default_mapper
self.default_renderer = default_renderer
[docs]def add_jsonrpc_endpoint(config, name, *args, **kw):
"""Add an endpoint for handling JSON-RPC.
``name``
The name of the endpoint.
``default_mapper``
A default view mapper that will be passed as the ``mapper``
argument to each of the endpoint's methods.
``default_renderer``
A default renderer that will be passed as the ``renderer``
argument to each of the endpoint's methods. This should be the
string name of the renderer, registered via
:meth:`pyramid.config.Configurator.add_renderer`.
A JSON-RPC method also accepts all of the arguments supplied to
:meth:`pyramid.config.Configurator.add_route`.
"""
default_mapper = kw.pop('default_mapper', MapplyViewMapper)
default_renderer = kw.pop('default_renderer', DEFAULT_RENDERER)
endpoint = Endpoint(
name,
default_mapper=default_mapper,
default_renderer=default_renderer,
)
config.registry.rpc_endpoints[name] = endpoint
predicates = kw.setdefault('custom_predicates', [])
predicates.append(EndpointPredicate())
config.add_route(name, *args, **kw)
config.add_view(exception_view, route_name=name, context=Exception,
permission=NO_PERMISSION_REQUIRED)
[docs]def add_jsonrpc_method(config, view, **kw):
"""Add a method to a JSON-RPC endpoint.
``endpoint``
The name of the endpoint.
``method``
The name of the method.
A JSON-RPC method also accepts all of the arguments supplied to
:meth:`pyramid.config.Configurator.add_view`.
A view mapper is registered by default which will match the
``request.rpc_args`` to parameters on the view. To override this
behavior simply set the ``mapper`` argument to None or another
view mapper.
.. note::
An endpoint **must** be defined before methods may be added.
"""
endpoint_name = kw.pop('endpoint', kw.pop('route_name', None))
if endpoint_name is None:
raise ConfigurationError(
'Cannot register a JSON-RPC endpoint without specifying the '
'name of the endpoint.')
endpoint = config.registry.rpc_endpoints.get(endpoint_name)
if endpoint is None:
raise ConfigurationError(
'Could not find an endpoint with the name "%s".' % endpoint_name)
# pop the method name
method = kw.pop('method', None)
if method is None:
raise ConfigurationError(
'Cannot register a JSON-RPC method without specifying the '
'"method"')
mapper = kw.pop('mapper', _marker)
if mapper is _marker:
# only override mapper if not supplied
mapper = endpoint.default_mapper
kw['mapper'] = mapper
renderer = kw.pop('renderer', None)
if renderer is None:
renderer = endpoint.default_renderer
kw['renderer'] = null_renderer
predicates = kw.setdefault('custom_predicates', [])
predicates.append(MethodPredicate(method, renderer=renderer))
decorator = kw.get('decorator', None)
if decorator is None:
decorator = jsonrpc_view
else:
if not is_nonstr_iter(decorator):
decorator = (decorator,)
# we want to apply the view_wrapper first, then the other decorators
# and combine() reverses the order, so ours goes last
decorators = list(decorator) + [jsonrpc_view]
decorator = combine(*decorators)
kw['decorator'] = decorator
config.add_view(view, route_name=endpoint_name, **kw)
[docs]class jsonrpc_method(object):
"""This decorator may be used with pyramid view callables to enable
them to respond to JSON-RPC method calls.
If ``method`` is not supplied, then the callable name will be used
for the method name.
This is the lazy analog to the
:func:`~pyramid_rpc.jsonrpc.add_jsonrpc_method`` and accepts all of
the same arguments.
"""
def __init__(self, method=None, **kw):
self.method = method
self.kw = kw
def __call__(self, wrapped):
kw = self.kw.copy()
kw['method'] = self.method or wrapped.__name__
def callback(context, name, ob):
config = context.config.with_package(info.module)
config.add_jsonrpc_method(view=ob, **kw)
info = venusian.attach(wrapped, callback, category='pyramid')
if info.scope == 'class':
# ensure that attr is set if decorating a class method
kw.setdefault('attr', wrapped.__name__)
kw['_info'] = info.codeinfo # fbo action_method
return wrapped
[docs]def includeme(config):
""" Set up standard configurator registrations. Use via:
.. code-block:: python
config = Configurator()
config.include('pyramid_rpc.jsonrpc')
Once this function has been invoked, two new directives will be
available on the configurator:
- ``add_jsonrpc_endpoint``: Add an endpoint for handling JSON-RPC.
- ``add_jsonrpc_method``: Add a method to a JSON-RPC endpoint.
"""
if not hasattr(config.registry, 'rpc_endpoints'):
config.registry.rpc_endpoints = {}
config.add_renderer(DEFAULT_RENDERER, jsonrpc_renderer)
config.add_directive('add_jsonrpc_endpoint', add_jsonrpc_endpoint)
config.add_directive('add_jsonrpc_method', add_jsonrpc_method)
config.add_view(exception_view, context=JsonRpcError,
permission=NO_PERMISSION_REQUIRED)
