Skip to content

aa.gusti / opensipkd-base

Go to a project
Commit 74734adf by aa.gustiana@gmail.com
Options

Implement JSON-RPC endpoint handling, enhance view configurations, and add API v… 

…iews for user registration and management
1 parent 422042ac
Inline Side-by-side
Showing 9 changed files with 322 additions and 114 deletions
opensipkd/base/__init__.py
......@@ -361,6 +361,7 @@ def _add_route(config, route):
if int(route.get("typ", 0)) == 0:
config.add_route(route.get("kode"), route.get("path"))
elif int(route.get("typ")) == 1:
config.add_jsonrpc_endpoint(route.get("kode"), route.get("path"),
default_renderer="json_rpc")
......@@ -399,10 +400,13 @@ def _add_view_config(config, paket, route):
params = dict(attr=f"{attr}",
route_name=route.get("kode"),
renderer=template)
if route.get("permission"):
params["permission"] = route.get("permission")
if route.get("crsf"):
params["require_csrf"] = True
if route.get("request_method"):
params["request_method"] = route.get("request_method")
config.add_view(views, **params)
......
opensipkd/base/scripts/data/routes.csv
......@@ -5,9 +5,9 @@ base-logout,/logout,base,user_login,ViewAuth,view_logout,1,,,Logout,1,0,,0,logou
base-password-reset,/password/reset,base,user_login,ViewPassword,reset_password,,,,Reset Password,1,0,,0,form6.pt,
base-password,/password,base,user_login,ViewPassword,change_password,1,view,,Change Password,1,0,,0,form8.pt,
base-password-request,/password/{code}/request,base,user_login,ViewPassword,change_password_request,1,,,Change Password,1,0,,0,form8.pt,
base-profile,/profile,base,register,,view_profile,,,,Profile,1,0,,0,form8.pt,
base-profile,/profile,base,register,,view_profile,,view,,Profile,1,0,,0,form8.pt,
base-register,/register,base,register,,view_register,,,,Register,1,0,,0,form8.pt,
base-recreate-api-key,/recreate-api-key,base,user_login,ViewPassword,recreate-api-key,,,,Get Api Key,1,0,,0,recreate-api-key.pt,
base-recreate-api-key,/recreate-api-key,base,register,ViewPassword,recreate-api-key,,,,Get Api Key,1,0,,0,recreate-api-key.pt,
base-admin,#,base,,,,,view,,Administrator,1,0,,1,,
base-user,/user,base,user,,view_list,,user-view,base-admin,User,1,0,,1,form.pt,
base-user-act,/user/{act}/act,base,user,,,,user-view,base-user,User Action,1,0,,,json,
......
opensipkd/base/views/api_base.py 0 → 100644
import datetime
from decimal import Decimal
from . import api_messages
from pyramid.response import Response
from pyramid.exceptions import HTTPNotFound
from opensipkd.base.models import DBSession
class ApiViews:
def __init__(self, request):
self.request = request
self.id = self.request.matchdict.get("id")
self.db_session = DBSession
self.table = None
self.pkey = ("id")
self.orders = None
self.psize = int(request.params.get("size", 25))
self.page = int(request.params.get("page", 1))
def filter_ids(self, query):
ids = self.id.split(".")
filters = dict(zip(self.pkey, ids))
return query.filter_by(**filters)
def get_orders(self, query):
if not self.orders:
self.orders = self.pkey
for k in self.orders:
query = query.order_by(getattr(self.table,k))
return query
def query(self, **kw):
table = kw.get("table", self.table)
filter_ids = kw.get("filters", self.filter_ids)
orders = kw.get("orders", self.get_orders)
query = self.db_session.query(table)
if self.id:
query = filter_ids(query)
else:
query = orders(query)
query = query.limit(self.psize).offset(
(self.page - 1) * self.psize)
return query
def success(self, data=[], msg=None):
result = dict(api_messages.SUCCESS)
result["data"] = data
result["error"]["msg"] = msg if msg else "Success"
return result
def get(self):
query=self.query()
if not query.first():
return HTTPNotFound()
data = []
for row in query:
d = dict(row.__dict__)
d.pop('_sa_instance_state', None)
for key, value in d.items():
if isinstance(value, datetime.datetime):
d[key] = value.isoformat()
elif isinstance(value, Decimal):
d[key] = float(value)
data.append(d)
return Response(json=self.success(data=data))
def post(self, data):
self.request = data
return self.request
def delete(self):
query = self.db_session.query(self.table)
query = self.filter_ids(query)
row = query.first()
if not row:
return HTTPNotFound()
return Response(json=self.success())
def put(self, data):
self.request = data
return self.request
def patch(self, data):
self.request = data
return self.request
opensipkd/base/views/api_messages.py 0 → 100644
SUCCESS = {
"error": {
"code": "0000",
"msg": "Data sudah dibayar"
}
}
ALLREADYPAID = {
"error": {
"code": "0001",
"msg": "Data sudah dibayar"
}
}
PAYMENTMISMATCH = {
"error": {
"code": "0002",
"msg": "Jumlah pembayaran tidak sesuai"
}
}
opensipkd/base/views/api_register.py 0 → 100644
from . import api_base
from ..models import Partner
class Views(api_base.ApiViews):
def __init__(self, request):
super().__init__(request)
self.table = Partner
def filter_ids(self, query):
return query.filter(Partner.email == self.request.user.email)
\ No newline at end of file
opensipkd/base/views/base_views.py
from decimal import Decimal
import logging
from math import tan
import os
import re
from datetime import datetime
from datetime import datetime, date
from email.utils import parseaddr
from cgi import FieldStorage
from webob.multidict import MultiDict
import colander
from datatables import ColumnDT
......@@ -199,26 +202,27 @@ class BaseView(object):
'tahun_akhir'] or self.tahun_akhir
self.ses['tahun_akhir'] = self.tahun_akhir
# self.departemen_kd = 'departemen_kd' in self.ses and self.ses[
# 'departemen_kd'] or '0.0.00'
# self.departemen_nm = 'departemen_nm' in self.ses and self.ses[
# 'departemen_nm'] or 'PILIH UNIT'
# self.departemen_id = 'departemen_id' in self.ses and self.ses[
# 'departemen_id'] or 0
# self.ses['departemen_kd'] = self.departemen_kd
# self.ses['departemen_nm'] = self.departemen_nm
# self.ses['departemen_id'] = self.departemen_id
# if 'departemen_id' in self.params:
# self.departemen_id = self.params['departemen_id']
# if not self.departemen_id:
# self.departemen_id = 0
# self.ses["departemen_id"] = self.departemen_id
# self.jenis = 'jenis' in self.ses and self.ses['jenis'] or 0
# self.jenis = 'jenis' in self.params and self.params[
# 'jenis'] or self.jenis
# self.ses['jenis'] = self.jenis
"""
self.departemen_kd = 'departemen_kd' in self.ses and self.ses[
'departemen_kd'] or '0.0.00'
self.departemen_nm = 'departemen_nm' in self.ses and self.ses[
'departemen_nm'] or 'PILIH UNIT'
self.departemen_id = 'departemen_id' in self.ses and self.ses[
'departemen_id'] or 0
self.ses['departemen_kd'] = self.departemen_kd
self.ses['departemen_nm'] = self.departemen_nm
self.ses['departemen_id'] = self.departemen_id
if 'departemen_id' in self.params:
self.departemen_id = self.params['departemen_id']
if not self.departemen_id:
self.departemen_id = 0
self.ses["departemen_id"] = self.departemen_id
self.jenis = 'jenis' in self.ses and self.ses['jenis'] or 0
self.jenis = 'jenis' in self.params and self.params[
'jenis'] or self.jenis
self.ses['jenis'] = self.jenis
"""
def form2dict(self, field):
children = []
for c in field.children:
......@@ -239,9 +243,11 @@ class BaseView(object):
}
return d
# def query_register(self, **kwargs):
# pass
"""
def query_register(self, **kwargs):
pass
"""
def get_routes(self):
"""
Digunakan untuk mendapatkan default url apabila list_url tidak ada
......@@ -274,26 +280,27 @@ class BaseView(object):
def form_validator(self, form, value):
pass
# def form_validate(self, form, err_value, **kwargs):
# controls = self.req.POST.items()
# try:
# c = form.validate(controls)
# except ValidationFailure as e:
# value = err_value()
# for f in e.field.children:
# if isinstance(f.typ, colander.Date):
# e.cstruct[f.name] = date_from_str(
# e.cstruct[f.name])
# if f.name == "captcha":
# e.cstruct[f.name] = self.get_captcha_url()
# value.update(e.cstruct)
# form.set_appstruct(e.cstruct)
# return self.returned_form(form, **kwargs)
# return dict(c)
# def get_params(self, params, default=None):
# return get_params(params, default)
"""
def form_validate(self, form, err_value, **kwargs):
controls = self.req.POST.items()
try:
c = form.validate(controls)
except ValidationFailure as e:
value = err_value()
for f in e.field.children:
if isinstance(f.typ, colander.Date):
e.cstruct[f.name] = date_from_str(
e.cstruct[f.name])
if f.name == "captcha":
e.cstruct[f.name] = self.get_captcha_url()
value.update(e.cstruct)
form.set_appstruct(e.cstruct)
return self.returned_form(form, **kwargs)
return dict(c)
def get_params(self, params, default=None):
return get_params(params, default)
"""
def get_form(self, class_form, row=None, buttons=(btn_save, btn_cancel),
**kwargs):
buttons = self.buttons and self.buttons or buttons
......@@ -326,11 +333,12 @@ class BaseView(object):
return Form(schema, buttons=buttons, autocomplete=self.autocomplete)
# def session_failed(self, session_name):
# r = dict(form=self.req.session[session_name])
# del self.req.session[session_name]
# return r
"""
def session_failed(self, session_name):
r = dict(form=self.req.session[session_name])
del self.req.session[session_name]
return r
"""
def view_list(self, **kwargs):
"""
custom:
......@@ -596,8 +604,7 @@ class BaseView(object):
def returned_form(self, form, **kwargs):
table = kwargs.get("table", None)
if self.req.is_xhr:
d = self.form2dict(form)
return Response(json=d["children"])
return self.resp_xhr({"data": [form.cstruct]})
resources = form.get_widget_resources()
readonly = "readonly" in kwargs and kwargs["readonly"] or False
......@@ -929,6 +936,28 @@ class BaseView(object):
def edit_restrict(self, row):
return False
def obj2json(self, values):
for key, val in values.items():
if isinstance(val, datetime):
values[key] = val.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(val, date):
values[key] = val.strftime('%Y-%m-%d')
elif isinstance(val, Decimal):
values[key] = float(val)
elif isinstance(val, colander._null):
values[key] = ""
return values
def resp_xhr(self, values):
if values.get("data"):
data = []
for val in values["data"]:
data.append(self.obj2json(val))
values["data"] = data
return Response(json=values)
def view_edit(self, **kwargs):
request = self.req
self.ses["readonly"] = False
......@@ -951,11 +980,28 @@ class BaseView(object):
if request.POST:
if 'save' in request.POST:
controls = request.POST.items()
log.debug(controls)
if self.req.is_xhr:
cloned = request.POST.items()
controls=[]
for ctrl in cloned:
if isinstance(ctrl[1], FieldStorage):
controls.append(("__start__", f"{ctrl[0]}:mapping"))
controls.append(("upload", ctrl[1]))
controls.append(("uid", ""))
controls.append(("__end__", f"{ctrl[0]}:mapping"))
log.debug(f"Control: {ctrl}")
else:
controls.append(ctrl)
items = MultiDict(controls)
controls = items.items()
log.debug(controls)
try:
controls = form.validate(controls)
controls = form.validate(controls)
except ValidationFailure as e:
log.error(f"Edit Error: {str(e.error.msg)}")
log.error(f"Edit Error: {str(e.error)}")
if self.req.is_xhr:
return self.resp_xhr({"error": e.error.asdict()})
for f in e.field.children:
if isinstance(f.typ, colander.Date):
e.cstruct[f.name] = date_from_str(
......@@ -969,15 +1015,18 @@ class BaseView(object):
c = dict(controls)
self.save_request(c, row)
if self.req.is_xhr:
form.set_appstruct(c)
d = self.form2dict(form)
return Response(json=d["children"])
return self.resp_xhr({"data": [c]})
return self.after_edit(row=row, **kwargs)
return self.next_edit(form, row=row)
# if self.req.is_xhr:
# return self.resp_xhr({"data": [form.cstruct]})
form.set_appstruct(values)
form = self.before_edit(form)
return self.returned_form(form, **kwargs)
......@@ -1128,6 +1177,7 @@ class BaseView(object):
def get_partner(self):
return Partner.query_email(self.req.user.email).first()
"""
# @colander.deferred
# def deferred_status(node, kw):
# values = kw.get('daftar_status', [])
......@@ -1170,3 +1220,4 @@ class BaseView(object):
# def get_url_captcha(request):
# captcha = get_captcha(request)
# return os.path.join(get_urls(request.route_url('home')), 'captcha', captcha)
"""
opensipkd/base/views/register.py
......@@ -116,15 +116,30 @@ class AddSchema(colander.Schema):
colander.String(),
widget=widget.TextInputWidget(),
title=_("ID Number"),
# missing=colander.drop,
oid="kode")
self["idcard"] = colander.SchemaNode(
FileData(),
widget=widget.FileUploadWidget(mem_tmp_store),
title=_("ID Card"),
missing=colander.drop,
validator=image_validator)
validator=image_validator)
# if request.is_xhr:
# if request.POST:
# self["idcard"] = colander.SchemaNode(
# FileData(),
# title=_("ID Card"),
# validator=image_validator)
# else:
# self["idcard"] = colander.SchemaNode(
# colander.String(),
# title=_("ID Card"),
# missing=colander.drop)
# else:
if BASE_CLASS.reg_nip:
self["nip"] = colander.SchemaNode(
colander.String(),
......@@ -313,6 +328,11 @@ class Views(BaseView):
user, value['password']):
err_login()
# if self.req.is_xhr:
# if "upload" in value and value["upload"]:
# value["idcard"] = value["upload"]
if "idcard" in value and value["idcard"]:
idcard = value["idcard"]
if "fp" in idcard and idcard["fp"] and idcard["fp"] != b'':
......@@ -418,20 +438,19 @@ class Views(BaseView):
"idcard"]
for f in fields:
d[f] = hasattr(partner, f) and getattr(partner, f) or ""
if "idcard" in d:
if d["idcard"]:
filename = d["idcard"]
preview_url = "/".join(
[self.req.static_url(BASE_CLASS.partner_doc),
filename])
d["idcard"] = {"uid": filename.split(".")[0],
"filename": filename,
"preview_url": preview_url
}
else:
d.pop("idcard")
else:
d.pop("idcard")
filename = d.get("idcard", "")
d.pop("idcard")
preview_url = "/".join(
[self.req.static_url(BASE_CLASS.partner_doc),
filename])
# if self.req.is_xhr: # Penambahan jika XHR tidak di parsing
# d["idcard"] = filename and preview_url or ""
# else:
if filename:
d["idcard"] = {"uid": filename.split(".")[0],
"filename": filename,
"preview_url": preview_url
}
return d
# def before_add(self):
......@@ -451,28 +470,30 @@ class Views(BaseView):
return resp
# def next_add(self, form, **kwargs):
# table = kwargs.get("table")
# kwargs.pop("table", None)
# resources = kwargs.get("resources")
# if 'register' in self.req.POST:
# controls = self.req.POST.items()
# try:
# c = form.validate(controls)
# except ValidationFailure as e:
# value = self.before_add()
# for f in e.field.children:
# if isinstance(f.typ, colander.Date):
# e.cstruct[f.name] = date_from_str(
# e.cstruct[f.name])
# if f.name == "captcha":
# e.cstruct[f.name] = self.get_captcha_url()
# value.update(e.cstruct)
# form.set_appstruct(e.cstruct)
# return self.returned_form(form, table, **kwargs)
# # js=resources["js"])
# values = dict(c)
# row = self.save_request(values)
# self.after_add(row=row, values=values)
# return self.route_list()
"""
def next_add(self, form, **kwargs):
table = kwargs.get("table")
kwargs.pop("table", None)
resources = kwargs.get("resources")
if 'register' in self.req.POST:
controls = self.req.POST.items()
try:
c = form.validate(controls)
except ValidationFailure as e:
value = self.before_add()
for f in e.field.children:
if isinstance(f.typ, colander.Date):
e.cstruct[f.name] = date_from_str(
e.cstruct[f.name])
if f.name == "captcha":
e.cstruct[f.name] = self.get_captcha_url()
value.update(e.cstruct)
form.set_appstruct(e.cstruct)
return self.returned_form(form, table, **kwargs)
# js=resources["js"])
values = dict(c)
row = self.save_request(values)
self.after_add(row=row, values=values)
return self.route_list()
"""
\ No newline at end of file
opensipkd/base/views/user_login.py
......@@ -47,7 +47,7 @@ from opensipkd.tools.buttons import btn_cancel
from .base_views import CSRFSchema, BaseView
from pyramid.i18n import TranslationStringFactory
from ..widgets import widget_os
import json
_ = TranslationStringFactory('login')
log = __import__("logging").getLogger(__name__)
......@@ -197,8 +197,10 @@ class ViewAuth(BaseView):
if request.authenticated_userid: # (request):
message = 'Anda sudah login'
if self.req.is_xhr:
return Response(json={"success": True,
"msg": message})
return Response(json={"error":
{"code": "0000",
"msg": message},
"data":[]})
request.session.flash('Anda sudah login', 'error')
return HTTPFound(location=f"{request.home}")
......@@ -224,8 +226,10 @@ class ViewAuth(BaseView):
msg = 'Login gagal'
set_user_log(msg, request, log, identity)
if self.req.is_xhr:
d = self.form2dict(e.field)
return Response(json=d["children"])
d = {"error": e.error.asdict()}
return Response(json=d)
# d = self.form2dict(e.field)
# return Response(json={"data": d["children"]})
request.session.flash(msg, 'error')
return HTTPFound(location=request.route_url('base-login'))
......@@ -255,8 +259,8 @@ class ViewAuth(BaseView):
request.session.flash(login.message, "error")
if self.req.is_xhr:
return Response(json={"error": {"code": -1,
"msg": login.message}
})
"msg": login.message},
"data":[]})
next_url = f"{request.route_url('base-login')}?next={next_url}"
return HTTPFound(location=next_url)
......@@ -280,7 +284,7 @@ class ViewAuth(BaseView):
if self.req.is_xhr:
# return Response(form.render())
d = self.form2dict(form)
return Response(json=d["children"])
return Response(json={"data": d["children"]})
return render_to_response(
login_tpl, dict(
form=form,
......@@ -303,10 +307,12 @@ class ViewAuth(BaseView):
# next_url=next_url,
# login=login, )
if self.req.is_xhr:
d = self.form2dict(form)
d = d["children"]
return self.resp_xhr({"data": [form.cstruct]})
# d = self.form2dict(form)
# d = d["children"]
# d["permission"]=user.get_permissions()
return Response(json=d)
return Response(json={"data": d})
if login_tpl:
return render_to_response(
......@@ -340,7 +346,8 @@ class ViewAuth(BaseView):
request.response.delete_cookie("g_state", '/')
if self.req.is_xhr:
return Response(json={"success": True,
"message": "Sukses Logout"},
"message": "Sukses Logout",
"data": []},
headerlist=headers)
form.set_appstruct({"message": "Sukses Logout"})
request.session["login"] = False
......@@ -359,8 +366,13 @@ def redirect_login(request, user):
if request.is_xhr:
return Response(json={
"success": True,
"permission": user.get_permissions(),
"token": user.security_code
"data": [
{
"permission": user.get_permissions(),
"token": user.security_code
}
],
}, headerlist=headers)
if not next_url and request.matched_route.name == 'login':
......
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!