Skip to content

irul / opensipkd-base

Go to a project
Commit 1f158947 by aagusti
Options

api

1 parent 1c9f3ec4
Expand all
Inline Side-by-side
Showing 15 changed files with 314 additions and 240 deletions
opensipkd/base/__init__.py
......@@ -66,29 +66,30 @@ titles = {}
# http://stackoverflow.com/questions/9845669/pyramid-inverse-to-add-notfound-viewappend-slash-true
class RemoveSlashNotFoundViewFactory(object):
def __init__(self, notfound_view=None):
if notfound_view is None:
notfound_view = default_exceptionresponse_view
self.notfound_view = notfound_view
def __call__(self, context, request):
if not isinstance(context, Exception):
# backwards compat for an append_notslash_view registered via
# config.set_notfound_view instead of as a proper exception view
context = getattr(request, 'exception', None) or context
path_req = request.path
registry = request.registry
mapper = registry.queryUtility(IRoutesMapper)
if mapper is not None and path_req.endswith('/'):
noslash_path = path_req.rstrip('/')
for route in mapper.get_routes():
if route.match(noslash_path) is not None:
qs = request.query_string
if qs:
noslash_path += '?' + qs
return HTTPFound(location=noslash_path)
return self.notfound_view(context, request)
# class RemoveSlashNotFoundViewFactory(object):
# diganti menggunakan @view_config(context=HTTPNotFound, renderer='templates/404.pt') pada base.views
# def __init__(self, notfound_view=None):
# if notfound_view is None:
# notfound_view = default_exceptionresponse_view
# self.notfound_view = notfound_view
#
# def __call__(self, context, request):
# if not isinstance(context, Exception):
# # backwards compat for an append_notslash_view registered via
# # config.set_notfound_view instead of as a proper exception view
# context = getattr(request, 'exception', None) or context
# path_req = request.path
# registry = request.registry
# mapper = registry.queryUtility(IRoutesMapper)
# if mapper is not None and path_req.endswith('/'):
# noslash_path = path_req.rstrip('/')
# for route in mapper.get_routes():
# if route.match(noslash_path) is not None:
# qs = request.query_string
# if qs:
# noslash_path += '?' + qs
# return HTTPFound(location=noslash_path)
# return self.notfound_view(context, request)
# https://groups.google.com/forum/#!topic/pylons-discuss/QIj4G82j04c
......@@ -312,11 +313,11 @@ def json_rpc():
return json_r
class MyAuthenticationPolicy(AuthTktAuthenticationPolicy):
def authenticated_userid(self, request):
user = request.user
if user is not None:
return user.id
# class MyAuthenticationPolicy(AuthTktAuthenticationPolicy):
# def authenticated_userid(self, request):
# user = request.user
# if user is not None:
# return user.id
def get_host(request):
......@@ -371,35 +372,21 @@ def main(global_config, **settings):
config = Configurator(settings=settings,
root_factory='opensipkd.base.models.RootFactory',
session_factory=session_factory)
from .models import RootFactory
modules = get_modules(settings)
# print(modules)
from importlib import import_module
for module in modules:
# compatibility
if module == 'admin':
continue
module = module.replace('/', '.')
mfile = module
print(">>Load Module:", mfile)
m = import_module(mfile)
cfg = m.main(config, **settings)
if cfg:
config = cfg
# todo apakah config bisa dikirim ke module?
# contoh:
# config = m.config(config)
# dipindahkan ke config pyramid.include
# config.include('pyramid_beaker')
# config.include('pyramid_chameleon')
# authn_policy = AuthTktAuthenticationPolicy(
# 'sosecret', callback=group_finder, hashalg='sha512')
#
# authz_policy = ACLAuthorizationPolicy()
config.set_security_policy(MySecurityPolicy(settings["session.secret"]))
# config.set_authentication_policy(authn_policy)
# config.set_security_policy(authz_policy)
# config.set_authorization_policy(authz_policy)
config.add_request_method(get_user, 'user', reify=True)
config.add_request_method(get_title, 'title', reify=True)
config.add_request_method(get_company, 'company', reify=True)
......@@ -421,70 +408,19 @@ def main(global_config, **settings):
config.add_request_method(allow_register, 'allow_register', reify=True)
config.add_request_method(disable_responsive, 'disable_responsive', reify=True)
config.add_request_method(get_params, 'get_params', reify=True)
# config.add_notfound_view(RemoveSlashNotFoundViewFactory())
config.add_static_view('static', 'opensipkd.base:static', cache_max_age=3600)
config.add_static_view('deform_static', 'deform:static')
# config.add_view('.views.api.echoGateway')
# config.add_static_view('files', get_params('static_files'))
# Captcha
captcha_files = get_params('captcha_files', settings=settings,alternate="/tmp/captcha")
captcha_files = get_params('captcha_files', settings=settings, alternate="/tmp/captcha")
if not os.path.exists(captcha_files):
os.makedirs(captcha_files)
config.add_static_view('captcha', captcha_files)
# config.add_static_view('tts', path=get_params('tts_files'))
config.add_static_view('captcha', captcha_files)
config.add_renderer('csv', 'opensipkd.tools.CSVRenderer')
config.add_renderer('json', json_renderer())
# dipindahkan ke config pyramid.include
# config.include('pyramid_rpc.jsonrpc')
config.add_renderer('json_rpc', json_rpc())
# q = DBSession.query(Route)
# for route in q:
# if route.type == 0:
# config.add_route(route.kode, route.path)
# if route.nama:
# titles[route.kode] = route.nama
# elif route.type == 1:
# config.add_jsonrpc_endpoint(route.kode, route.path,
# default_renderer="json_rpc")
set_routes(config)
###########################################
# MAP
# todo apabila config bosa di get dari module maka baris ini bisa hilang
# Sudah solve menggunakan includeme
###########################################
# if 'opensipkd.map.base' in modules:
# import papyrus
# from papyrus.renderers import GeoJSON, XSD
#
# config.add_request_method(get_gmap_key, 'gmap_key', reify=True)
# config.add_request_method(get_bing_key, 'bing_key', reify=True)
# config.add_request_method(get_extent, 'extent', reify=True)
#
# config.include(papyrus.includeme)
# config.add_renderer('geojson', GeoJSON())
# config.add_renderer('xsd', XSD())
# config.add_static_view('static_map', 'opensipkd.map.base:static', cache_max_age=3600)
# if 'opensipkd.map.aset' in modules:
# config.add_static_view('static_map_aset', 'opensipkd.map.aset:static', cache_max_age=3600)
#
# # if 'opensipkd.map.pbb' in modules:
# # config.add_static_view('static_map_pbb', 'opensipkd.map.pbb:static', cache_max_age=3600)
#
# if 'opensipkd.pasar.web' in modules:
# config.add_static_view('static_pasar', 'opensipkd.pasar.web:static', cache_max_age=3600)
#
# if 'opensipkd.pbb.master' in modules:
# config.add_static_view('static_pbb', 'opensipkd.pbb.master:static', cache_max_age=3600)
# if 'opensipkd.pos.pbb' in modules:
# config.add_static_view('static_pospbb', 'opensipkd.pos.pbb:static', cache_max_age=3600)
config.registry['mailer'] = mailer_factory_from_settings(settings)
# config.include()
config.scan()
for m in modules:
config.scan(m)
......
opensipkd/base/alembic/versions/8e7057155823_penambahan_user_device_expired.py 0 → 100644
"""penambahan user device expired
Revision ID: 8e7057155823
Revises: 86c1b4a1da16
Create Date: 2022-07-08 14:58:39.378811
"""
# revision identifiers, used by Alembic.
revision = '8e7057155823'
down_revision = '86c1b4a1da16'
branch_labels = None
depends_on = None
from alembic import op
import sqlalchemy as sa
def upgrade():
context = op.get_context()
helpers = context.opts['helpers']
if not helpers.table_has_column('user_device', 'expired'):
op.add_column('user_device', sa.Column('expired', sa.DateTime(timezone=True)))
op.alter_column('user_device', sa.Column('expired', sa.DateTime(timezone=True)))
if not helpers.table_has_column('routes', 'create_uid'):
op.add_column('routes', sa.Column('create_uid', sa.Integer, default=0))
if not helpers.table_has_column('routes', 'update_uid'):
op.add_column('routes', sa.Column('update_uid', sa.Integer, default=0))
def downgrade():
pass
opensipkd/base/models/common.py
......@@ -38,8 +38,9 @@ class UserDeviceModel(Base, KodeModel):
kode = Column(String(256))
token = Column(String(256))
logged_in = Column(Integer)
las_login_date = Column(DateTime)
las_login_date = Column(DateTime(timezone=True))
expired = Column(DateTime(timezone=True))
user = relationship(User, backref=backref("devices"))
class ResCompany(Base, NamaModel):
__tablename__ = 'company'
......
opensipkd/base/scripts/initializedb.py
......@@ -258,14 +258,16 @@ def alembic_run(ini_file, name=None):
if subprocess.call(command) != 0:
sys.exit()
def base_alembic_run(ini_file, name=None):
def alembic_run(ini_file, name='alembic_base'):
bin_path = os.path.split(sys.executable)[0]
alembic_bin = os.path.join(bin_path, 'alembic')
command = (alembic_bin, '-c', ini_file, '-n', 'alembic_base', 'upgrade', 'head')
command = (alembic_bin, '-c', ini_file, '-n', name, 'upgrade', 'head')
if subprocess.call(command) != 0:
sys.exit()
def base_alembic_run(ini_file):
alembic_run(ini_file)
def main(argv=sys.argv):
if len(argv) < 2:
......
opensipkd/base/security.py
import logging
from opensipkd.base.tools.api import rpc_auth
from opensipkd.tools.api import JsonRpcInvalidLoginError
from pyramid.renderers import render_to_response
from .models import (
User,
UserGroup,
......@@ -35,6 +40,7 @@ def get_user(request):
q = DBSession.query(User).filter_by(id=user_id)
return q.first()
# def get_user(request):
# user_id = request.unauthenticated_userid
# if user_id is not None:
......@@ -45,21 +51,18 @@ def get_user(request):
from pyramid.authentication import AuthTktCookieHelper
from pyramid.authorization import ACLHelper, Authenticated, Everyone
class MySecurityPolicy:
def __init__(self, secret):
self.helper = AuthTktCookieHelper(secret)
def identity(self, request):
# define our simple identity as None or a dict with userid and principals keys
identity = self.helper.identify(request)
if identity is None:
return None
userid = identity['userid'] # identical to the deprecated request.unauthenticated_userid
# verify the userid, just like we did before with groupfinder
userid = identity['userid']
principals = group_finder(userid, request)
# assuming the userid is valid, return a map with userid and principals
if principals is not None:
return {
'userid': userid,
......@@ -67,18 +70,14 @@ class MySecurityPolicy:
}
def authenticated_userid(self, request):
# defer to the identity logic to determine if the user id logged in
# and return None if they are not
identity = request.identity
if identity is not None:
return identity['userid']
def permits(self, request, context, permission):
# use the identity to build a list of principals, and pass them
# to the ACLHelper to determine allowed/denied
identity = request.identity
principals = set([Everyone])
if identity is not None:
principals.add(Authenticated)
principals.add(identity['userid'])
......@@ -90,4 +89,3 @@ class MySecurityPolicy:
def forget(self, request, **kw):
return self.helper.forget(request, **kw)
opensipkd/base/tools/__init__.py
from .. import log
import logging
log=logging.getLogger(__name__)
log.warning("opensipkd.base.tools depreciated use opensipkd.tools")
from opensipkd.tools import *
opensipkd/base/tools/api.py
import json
from datetime import timedelta, timezone, tzinfo
import requests
from opensipkd.tools import (
get_random_number, devel, get_random_string, get_settings)
get_random_number, devel, get_random_string, get_settings, DefaultTimeZone, get_params, get_timezone)
from opensipkd.tools.api import *
from .. import log
from ..models import (DBSession, User, GroupPermission, UserDeviceModel)
import logging
lima_menit = 300
log = logging.getLogger(__name__)
lima_menit = 300
#
def auth_from_rpc(request):
return auth_from(request)
def rpc_auth(request):
return auth_from(request)
def auth_from(request, field=None):
global lima_menit
......@@ -43,27 +50,44 @@ def auth_from(request, field=None):
return user
def auth_from_token(request):
return auth_from(request, "security_code")
def renew_token(user_device):
user_device.token = get_random_string(32)
DBSession.add(user_device)
DBSession.flush()
return user_device
def get_user_device(request, user):
# def auth_from_token(request):
# return auth_from(request, "security_code")
#
# def renew_token(user_device, logout=False):
# now = datetime.now(tz=get_timezone())
# tte = timedelta(minutes=10)
# if not user_device.expired or not user_device.token or \
# now - user_device.expired > tte:
# user_device.expired = now
# user_device.token = get_random_string(128)
# if logout:
# user_device.token=""
# DBSession.add(user_device)
# DBSession.flush()
# return user_device
# def token_auth(request, logout=False):
# if not request.environ["HTTP_TOKEN"]:
# raise JsonRpcInvalidLoginError
# user_device = UserDeviceModel.query() \
# .filter_by(kode=request.environ["HTTP_USER_AGENT"],
# token=request.environ["HTTP_TOKEN"]).first()
# if not user_device:
# raise JsonRpcInvalidLoginError
#
# return renew_token(user_device, logout=logout)
def get_user_device(request, user_id):
user_device = UserDeviceModel.query() \
.filter_by(user_id=user.id,
.filter_by(user_id=user_id,
kode=request.environ["HTTP_USER_AGENT"]).first()
if not user_device:
user_device = UserDeviceModel()
user_device.user_id = user.id
user_device.user_id = user_id
user_device.kode = request.environ["HTTP_USER_AGENT"]
user_device.token = get_random_string(32)
DBSession.add(user_device)
DBSession.flush()
# user_device = renew_token(user_device)
return user_device
......@@ -82,6 +106,7 @@ def validate_time(request):
return time_stamp
def auth_device(request):
env = request.environ
log.info(env)
......
opensipkd/base/views/__init__.py
......@@ -4,15 +4,20 @@ from datetime import timedelta
import colander
from deform import (
Form, ValidationFailure, widget, Button, )
from opensipkd.tools.api import JsonRpcInvalidLoginError
from pyramid.httpexceptions import (
HTTPFound, HTTPForbidden, HTTPNotFound, HTTPInternalServerError,
HTTPSeeOther)
from pyramid.i18n import TranslationStringFactory
from pyramid.interfaces import IRoutesMapper
from pyramid.renderers import render_to_response
from pyramid.response import Response
from pyramid.security import remember
from pyramid.view import view_config
from opensipkd.base import get_params
from opensipkd.base.tools.api import rpc_auth
from .base_views import BaseView
from ..models import (
DBSession, UserService, )
......@@ -74,18 +79,11 @@ class Home(BaseView):
@view_config(context=HTTPForbidden, renderer='templates/403.pt')
def http_forbidden(request):
# if request.authenticated_userid: # (request):
# request.session.flash('Hak Akses Terbatas', 'error')
# return HTTPFound(location=request.route_url('home'))
# return HTTPFound(location=request.route_url('login'))
if not request.is_authenticated:
next_url = request.route_url('login', _query={'next': request.url})
# next_url = f'{get_params("_host").strip()}/{next_url}'
# log.info(next_url)
return HTTPSeeOther(location=next_url)
request.response.status = 403
request.response.status = 403
return {"url": request.url}
......
opensipkd/base/views/api.py
......@@ -6,35 +6,7 @@ from ziggurat_foundations.models.services.user import UserService
from ..models import Partner, User
def get_profile_(user):
partner = Partner.query().filter_by(email=user.email).first()
if not partner:
raise JsonRpcInvalidDataError
result = dict(user_name=user.user_name,
nik=partner.kode,
email=partner.email,
mobile=partner.mobile,
nama=partner.nama, )
return dict(data=result)
@jsonrpc_method(method='get_profile', endpoint='rpc-user')
def get_profile(request, data):
"""
Digunakan untuk memperoleh profile user yang sedang login
parameter
@param request: Request
@param data: Dict(user_name=user_name/email, password=password)
@return:
"""
user = request.user
print("User", user)
data = type(data) == list and data[0] or data
user = User.get_by_identity(data["user_name"])
if not user or not UserService.check_password(user, data['password']):
raise JsonRpcInvalidLoginError
return get_profile_(user)
# services = {
......
opensipkd/base/views/partner_base.py
......@@ -12,12 +12,13 @@ class NamaSchema(colander.Schema):
validator=colander.Length(max=32),
oid="kode",
title="Kode",
width="100pt")
width="100pt")
nama = colander.SchemaNode(
colander.String(),
validator=colander.Length(max=64),
oid="nama")
class PartnerSchema(NamaSchema):
alamat_1 = colander.SchemaNode(
colander.String(),
......
opensipkd/base/views/templates/_table.pt
<html metal:use-macro="load: ./base3.1.pt">
<html metal:use-macro="load: ./base.pt">
<js metal:fill-slot="js_files">
<script src="${home}/static/v3/js/plugin/datatables/jquery.dataTables.min.js"></script>
<script src="${home}/static/v3/js/plugin/datatables/jquery.dataTables.min.js"></script>
<script src="${home}/static/v3/js/plugin/datatables/dataTables.colVis.min.js"></script>
<script src="${home}/static/v3/js/plugin/datatables/dataTables.tableTools.min.js"></script>
<script src="${home}/static/v3/js/plugin/datatables/dataTables.bootstrap.min.js"></script>
<script src="${home}/static/v3/js/plugin/datatable-responsive/datatables.responsive.min.js"></script>
</js>
</html>
</html>
\ No newline at end of file
opensipkd/base/views/user.py
......@@ -94,8 +94,8 @@ class Views(BaseView):
base_path=base_path)
filename = os.path.basename(filename)
resp = pdf_response(self.req, pdf, filename)
if resp.content_length<10:
resp.content_length=len(resp.body)
if resp.content_length < 10:
resp.content_length = len(resp.body)
return resp
return super(Views, self).view_act()
......@@ -238,21 +238,22 @@ class EmailValidator(colander.Email, Validator):
Validator.__init__(self, user)
def __call__(self, node, value):
def email_found():
data = dict(email=email, uid=found.id)
ts = _(
'email-already-used',
default='Email ${email} already used by user ID ${uid}',
mapping=data)
raise colander.Invalid(node, ts)
if self.match_object.match(value) is None:
raise colander.Invalid(node, _('Invalid email format'))
email = value.lower()
if self.user and self.user.email == email:
return
q = DBSession.query(User).filter_by(email=email)
found = q.first()
if not found:
return
data = dict(email=email, uid=found.id)
ts = _(
'email-already-used',
default='Email ${email} already used by user ID ${uid}',
mapping=data)
raise colander.Invalid(node, ts)
if found and (not self.user or self.user.email!=found.email):
email_found()
REGEX_ONLY_CONTAIN = re.compile('([A-Za-z0-9-]*)')
......
opensipkd/jsonrpc_auth/__init__.py
import logging
import venusian
from pyramid.httpexceptions import HTTPFound
from pyramid.httpexceptions import HTTPForbidden
from pyramid.httpexceptions import HTTPNotFound
from pyramid.renderers import null_renderer
from pyramid.security import NO_PERMISSION_REQUIRED
from pyramid_rpc.jsonrpc import jsonrpc_method as json_rpc_base, MethodPredicate, BatchedRequestPredicate, \
EndpointPredicate, jsonrpc_renderer, DEFAULT_RENDERER, add_jsonrpc_endpoint, add_jsonrpc_method, JsonRpcError, \
exception_view, JsonRpcRequestInvalid, parse_request_GET, parse_request_POST
from opensipkd.base.tools.api import auth_from_rpc
from opensipkd.base.views.user_login import get_login_headers
from pyramid_rpc.jsonrpc import (JsonRpcError, JsonRpcMethodNotFound, JsonRpcParamsInvalid,
JsonRpcInternalError, make_error_response, MethodPredicate, BatchedRequestPredicate,
jsonrpc_renderer, add_jsonrpc_method,
DEFAULT_RENDERER,
batched_request_view, Endpoint, EndpointPredicate)
from pyramid_rpc.mapper import ViewMapperArgsInvalid, MapplyViewMapper
log = logging.getLogger(__name__)
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)
class JsonRpcRequestForbidden(JsonRpcError):
code = -32604
message = 'request forbidden'
class JsonRpcInvalidLogin(JsonRpcError):
code = -32605
message = "Invalid User/Password"
#
# class EndpointPredicate(BaseEndpointPredicate):
# def __call__(self, info, request):
# if self.val:
# # find the endpoint info
# key = info['route'].name
# endpoint = request.registry.jsonrpc_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
# 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 hasattr(request, 'batched_rpc_requests'):
# log.debug('handling batched rpc request')
# # the checks below will look at the subrequests
# return
#
# 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
# env = request.environ
# if 'HTTP_TOKEN' in env:
# try:
# user_device = token_auth(request)
# user = user_device.user
# headers = remember(request, user.id)
# request.headers["Cookie"] = dict(headers)["Set-Cookie"]
# request.headers["token"]=user_device.token
# log.debug(request.headers["Cookie"])
# except JsonRpcInvalidLoginError as e:
# raise JsonRpcInvalidLogin
#
# elif ('HTTP_USERID' in env and 'HTTP_SIGNATURE' in env and
# 'HTTP_KEY' in env):
# try:
# user = rpc_auth(request)
# headers = remember(request, user.id)
# request.headers["Cookie"] = dict(headers)["Set-Cookie"]
# log.debug(request.headers["Cookie"])
# except JsonRpcInvalidLoginError as e:
# raise JsonRpcInvalidLogin
# log.debug('handling id:%s method:%s',
# request.rpc_id, request.rpc_method)
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 = JsonRpcRequestForbidden()
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:
log.debug('unsupported request method "%s"', request.method)
raise JsonRpcRequestInvalid
if hasattr(request, 'batched_rpc_requests'):
log.debug('handling batched rpc request')
# the checks below will look at the subrequests
return
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)
class MethodPredicate(object):
def __init__(self, val, config):
self.method = val
def text(self):
return 'jsonrpc method = %s' % self.method
phash = text
def __call__(self, context, request):
user = auth_from_rpc(request)
headers = get_login_headers(request, user)
response = HTTPFound(location=request.route_url('home'), headers=headers)
# response = request.response
request.response.set_cookie('userid', value=str(user.id), max_age=31536000) # max_age = year
return getattr(request, 'rpc_method', None) == self.method
fault = JsonRpcInternalError()
log.exception('json-rpc exception rpc_id:%s "%s"', rpc_id, exc)
return make_error_response(request, fault, rpc_id)
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.jsonrpc_endpoints[name] = endpoint
kw['jsonrpc_endpoint'] = True
config.add_route(name, *args, **kw)
kw = {}
kw['jsonrpc_batched'] = True
kw['renderer'] = null_renderer
config.add_view(batched_request_view, route_name=name,
permission=NO_PERMISSION_REQUIRED, **kw)
config.add_view(exception_view, route_name=name, context=Exception,
permission=NO_PERMISSION_REQUIRED)
def includeme(config):
""" Set up standard configurator registrations. Use via:
......
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!