You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
cms/cms/__init__.py

186 lines
6.6 KiB
Python

from genericpath import isdir
import os
import pandas as pd
import copy
from jinja2 import Environment, BaseLoader, FileSystemLoader
import importlib
import importlib.util
import json
from . import sites
from . import apexchart
from . import meta
class Plugin :
#
# decorator for plugin functions, this is a preliminary to enable future developement
#
def __init__(self,**_args):
self._mimetype = _args['mimetype']
if 'method' in _args :
_method = _args['method']
if type(_method) != list :
_method = [_method]
else:
_method = ['POST','GET']
self._method = _method
def __call__(self,_callback):
def wrapper(**_args):
return _callback(**_args)
setattr(wrapper,'method',self._method)
setattr(wrapper,'mimetype',self._mimetype)
return wrapper
# @staticmethod
# def load(_path,_filename,_name) :
# """
# This function will load external module form a given location and return a pointer to a function in a given module
# :path absolute path of the file (considered plugin) to be loaded
# :name name of the function to be applied
# """
# # _path = _args['path'] #os.sep.join([_args['root'],'plugin'])
# if os.path.isdir(_path):
# files = os.listdir(_path)
# if files :
# files = [name for name in files if name.endswith('.py') and name == _filename]
# if files:
# _path = os.sep.join([_path,files[0]])
# else:
# return None
# else:
# return None
# #-- We have a file ...
# # _name = _args['name']
# spec = importlib.util.spec_from_file_location(_filename, _path)
# module = importlib.util.module_from_spec(spec)
# spec.loader.exec_module(module)
# #
# # we need to make sure we have the plugin decorator here to make sure
# return getattr(module,_name) if hasattr(module,_name) else None
# @staticmethod
# def stats (_config) :
# """
# Returns the statistics of the plugins
# """
# _data = []
# for _name in _config :
# _log = {"files":_name,"loaded":len(_config[_name]),"logs":json.dumps(_config[_name])}
# _data.append(_log)
# return pd.DataFrame(_data)
@staticmethod
def call(**_args):
"""
This function will execute a plugged-in function given
"""
# _uri = _args['uri'] if 'uri' in _args else '/'.join([_args['module'],_args['name']])
_handler= _args['handler'] #-- handler
_request= _args['request']
_uri = _args['uri'] #_request.path[1:]
#
# we need to update the _uri (if context/embeded apps)
#
_context = _handler.system()['context']
if _context :
_uri = f'{_context}/{_uri}'
_plugins = _handler.plugins()
_code = 200
if _uri in _plugins :
_config = _handler.config() #_args['config']
_pointer = _plugins[_uri]
if hasattr(_pointer,'mimetype') and _request.method in _pointer.method:
#
# we constraint the methods given their execution ...
_mimeType = _pointer.mimetype
_data = _pointer(request=_request,config=_config)
else:
_mimeType = 'application/octet-stream'
try:
_data,_mimeType = _pointer(request=_request,config=_config)
except Exception as e:
_data = _pointer(request=_request,config=_config)
if type(_data) == pd.DataFrame :
_data = _data.to_json(orient='records')
elif type(_data) in [dict,list] :
_data = json.dumps(_data)
pass
else:
_code = 404
#
# We should generate a 500 error in this case with a message ...
#
_mimeType = 'plain/html'
_data = f"""
<script>
</script>
"""
return _data,_code,{'Content-Type':_mimeType}
pass
class delegate:
def __call__(self,**_args) :
_handler= _args['handler'] #-- module handler
_request= _args['request']
_uri = _args['uri'] #_request.path[1:]
#
# we need to update the _uri (if context/embeded apps)
#
# _context = _handler.system()['context']
_context = _handler.get('system.context')
if _context :
_uri = f'{_context}/{_uri}'
# print ([' ** ',_uri, _args['uri']])
# _plugins = _handler.plugins()
_plugins = _handler.get('plugins')
_code = 200
if _uri in _plugins :
# _config = _handler.config() #_args['config']
_config = _handler.get(None) #_args['config']
_pointer = _plugins[_uri]
_data = {}
if hasattr(_pointer,'mimetype') and hasattr(_pointer,'method'):
#
# we constraint the methods given their execution ...
_mimeType = _pointer.mimetype
if _request.method in _pointer.method :
_data = _pointer(request=_request,config=_config)
elif not hasattr(_pointer,'mimetype'):
_data,_mimeType = _pointer(request=_request,config=_config)
else:
_mimeType = 'application/octet-stream'
# _data = _pointer(request=_request,config=_config)
if type(_data) == pd.DataFrame :
_data = _data.to_json(orient='records')
elif type(_data) in [dict,list] :
_data = json.dumps(_data)
pass
else:
_code = 404
#
# We should generate a 500 error in this case with a message ...
#
_mimeType = 'plain/html'
_data = f"""
<script>
//
// I should
_msg = '<div class="border-round" style="display:grid; grid-template-columns:100px auto; gap:8px; align-content:center"><div class="large-text">404</div>,div> Resource NOT found</di><div>'
</script>
"""
return _data,_code,{'Content-Type':_mimeType}