Automated update of Base Application package.

Browse Source
This commit is contained in:
bot 2024-12-20 22:43:03 +00:00
parent 0f1495c5bd
commit cd450dba57
3 changed files with 77 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
dist/app-1.0.0-py3-none-any.whl vendored
View File

Binary file not shown.

BIN
dist/app-1.0.0.tar.gz vendored
View File

Binary file not shown.

151
src/app/rpc.py
View File

@ -1,31 +1,37 @@
from xmlrpc.server import resolve_dotted_attribute
from xmlrpc.client import Fault, dumps, loads, gzip_encode, gzip_decode, ServerProxy,MultiCall
from functools import partial
from inspect import signature
from aiohttp import web
from datetime import datetime from datetime import datetime
from functools import partial
from xmlrpc.client import (
Fault,
MultiCall,
ServerProxy,
dumps,
loads,
)
from xmlrpc.server import resolve_dotted_attribute
from aiohttp import web
class AsyncSimpleXMLRPCDispatcher: class AsyncSimpleXMLRPCDispatcher:
""" """
Original not async version of this class is in the original python std lib: Original not async version of this class is in the original python std lib:
https://github.com/python/cpython/blob/main/Lib/xmlrpc/server.py. https://github.com/python/cpython/blob/main/Lib/xmlrpc/server.py.
use_builtin_types=True allows the use of bytes-object which is preferred use_builtin_types=True allows the use of bytes-object which is preferred
because else it's a custom xmlrpc.client.Binary which sucks. because else it's a custom xmlrpc.client.Binary which sucks.
""" """
def __init__(self, instance,allow_none=True, encoding="utf-8", use_builtin_types=True): def __init__(
self, instance, allow_none=True, encoding="utf-8", use_builtin_types=True
):
self.setup_rpc(allow_none=allow_none, encoding=encoding, use_builtin_types=True) self.setup_rpc(allow_none=allow_none, encoding=encoding, use_builtin_types=True)
self.register_instance(instance,True) self.register_instance(instance, True)
def setup_rpc(self, allow_none=True, encoding="utf-8", def setup_rpc(self, allow_none=True, encoding="utf-8", use_builtin_types=True):
use_builtin_types=True):
self.funcs = {} self.funcs = {}
self.instance = None self.instance = None
self.allow_none = allow_none self.allow_none = allow_none
self.encoding = encoding or 'utf-8' self.encoding = encoding or "utf-8"
self.use_builtin_types = use_builtin_types self.use_builtin_types = use_builtin_types
def register_instance(self, instance, allow_dotted_names=True): def register_instance(self, instance, allow_dotted_names=True):
@ -45,14 +51,18 @@ class AsyncSimpleXMLRPCDispatcher:
return function return function
def register_introspection_functions(self): def register_introspection_functions(self):
self.funcs.update({'system.listMethods' : self.system_listMethods, self.funcs.update(
'system.methodSignature' : self.system_methodSignature, {
'system.methodHelp' : self.system_methodHelp}) "system.listMethods": self.system_listMethods,
"system.methodSignature": self.system_methodSignature,
"system.methodHelp": self.system_methodHelp,
}
)
def register_multicall_functions(self): def register_multicall_functions(self):
self.funcs.update({'system.multicall' : self.system_multicall}) self.funcs.update({"system.multicall": self.system_multicall})
async def _marshaled_dispatch(self, data, dispatch_method = None, path = None): async def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
try: try:
params, method = loads(data, use_builtin_types=self.use_builtin_types) params, method = loads(data, use_builtin_types=self.use_builtin_types)
@ -61,45 +71,47 @@ class AsyncSimpleXMLRPCDispatcher:
else: else:
response = await self._dispatch(method, params) response = await self._dispatch(method, params)
response = (response,) response = (response,)
response = dumps(response, methodresponse=1, response = dumps(
allow_none=self.allow_none, encoding=self.encoding) response,
methodresponse=1,
allow_none=self.allow_none,
encoding=self.encoding,
)
except Fault as fault: except Fault as fault:
response = dumps(fault, allow_none=self.allow_none, response = dumps(fault, allow_none=self.allow_none, encoding=self.encoding)
encoding=self.encoding)
except BaseException as exc: except BaseException as exc:
response = dumps( response = dumps(
Fault(1, "%s:%s" % (type(exc), exc)), Fault(1, f"{type(exc)}:{exc}"),
encoding=self.encoding, allow_none=self.allow_none, encoding=self.encoding,
) allow_none=self.allow_none,
)
return response.encode(self.encoding, 'xmlcharrefreplace') return response.encode(self.encoding, "xmlcharrefreplace")
def system_listMethods(self): def system_listMethods(self):
methods = set(self.funcs.keys()) methods = set(self.funcs.keys())
if self.instance is not None: if self.instance is not None:
if hasattr(self.instance, '_listMethods'): if hasattr(self.instance, "_listMethods"):
methods |= set(self.instance._listMethods()) methods |= set(self.instance._listMethods())
elif not hasattr(self.instance, '_dispatch'): elif not hasattr(self.instance, "_dispatch"):
methods |= set(list_public_methods(self.instance)) methods |= set(list_public_methods(self.instance))
return sorted(methods) return sorted(methods)
def system_methodSignature(self, method_name): def system_methodSignature(self, method_name):
return 'signatures not supported' return "signatures not supported"
def system_methodHelp(self, method_name): def system_methodHelp(self, method_name):
method = None method = None
if method_name in self.funcs: if method_name in self.funcs:
method = self.funcs[method_name] method = self.funcs[method_name]
elif self.instance is not None: elif self.instance is not None:
if hasattr(self.instance, '_methodHelp'): if hasattr(self.instance, "_methodHelp"):
return self.instance._methodHelp(method_name) return self.instance._methodHelp(method_name)
elif not hasattr(self.instance, '_dispatch'): elif not hasattr(self.instance, "_dispatch"):
try: try:
method = resolve_dotted_attribute( method = resolve_dotted_attribute(
self.instance, self.instance, method_name, self.allow_dotted_names
method_name, )
self.allow_dotted_names
)
except AttributeError: except AttributeError:
pass pass
if method is None: if method is None:
@ -110,21 +122,17 @@ class AsyncSimpleXMLRPCDispatcher:
async def system_multicall(self, call_list): async def system_multicall(self, call_list):
results = [] results = []
for call in call_list: for call in call_list:
method_name = call['methodName'] method_name = call["methodName"]
params = call['params'] params = call["params"]
try: try:
results.append([await self._dispatch(method_name, params)]) results.append([await self._dispatch(method_name, params)])
except Fault as fault: except Fault as fault:
results.append( results.append(
{'faultCode' : fault.faultCode, {"faultCode": fault.faultCode, "faultString": fault.faultString}
'faultString' : fault.faultString} )
)
except BaseException as exc: except BaseException as exc:
results.append( results.append({"faultCode": 1, "faultString": f"{type(exc)}:{exc}"})
{'faultCode' : 1,
'faultString' : "%s:%s" % (type(exc), exc)}
)
return results return results
async def _dispatch(self, method, params): async def _dispatch(self, method, params):
@ -135,17 +143,15 @@ class AsyncSimpleXMLRPCDispatcher:
else: else:
if func is not None: if func is not None:
return await func(*params) return await func(*params)
raise Exception('method "%s" is not supported' % method) raise Exception(f'method "{method}" is not supported')
if self.instance is not None: if self.instance is not None:
if hasattr(self.instance, '_dispatch'): if hasattr(self.instance, "_dispatch"):
return await self.instance._dispatch(method, params) return await self.instance._dispatch(method, params)
try: try:
func = resolve_dotted_attribute( func = resolve_dotted_attribute(
self.instance, self.instance, method, self.allow_dotted_names
method,
self.allow_dotted_names
) )
except AttributeError: except AttributeError:
pass pass
@ -153,20 +159,21 @@ class AsyncSimpleXMLRPCDispatcher:
if func is not None: if func is not None:
return await func(*params) return await func(*params)
raise Exception('method "%s" is not supported' % method) raise Exception(f'method "{method}" is not supported')
def rpc_wrap_instance(obj): def rpc_wrap_instance(obj):
class Session: class Session:
def __init__(self,data=None): def __init__(self, data=None):
self._data = data or {} self._data = data or {}
async def get(self, key, default=None): async def get(self, key, default=None):
return self._data.get(key,default) return self._data.get(key, default)
async def set(self,key, value): async def set(self, key, value):
self._data[key] = value self._data[key] = value
async def delete(self, key): async def delete(self, key):
try: try:
@ -175,39 +182,35 @@ def rpc_wrap_instance(obj):
except KeyError: except KeyError:
return False return False
async def exists(self,key): async def exists(self, key):
return key in self._data return key in self._data
class Instance: class Instance:
def __init__(self, _self):
self._self = self
self.session = Session()
def __get__(self,key):
return getattr(self._self,key)
def ping(self,*args,**kwargs): def __init__(self, _self):
return dict( self._self = self
args=args, self.session = Session()
kwargs=kwargs,
timestamp=str(datetime.now()) def __get__(self, key):
) return getattr(self._self, key)
def ping(self, *args, **kwargs):
return {"args": args, "kwargs": kwargs, "timestamp": str(datetime.now())}
instance = Instance(obj) instance = Instance(obj)
for attr in dir(obj): for attr in dir(obj):
if attr == 'rpc_handler': if attr == "rpc_handler":
continue continue
if attr.startswith("rpc_") and callable(getattr(obj, attr)): if attr.startswith("rpc_") and callable(getattr(obj, attr)):
setattr(instance,attr[4:], getattr(obj,attr)) setattr(instance, attr[4:], getattr(obj, attr))
return instance return instance
class Application(web.Application): class Application(web.Application):
def __init__(self, url=None,host=None,port=None, *args, **kwargs): def __init__(self, url=None, host=None, port=None, *args, **kwargs):
self.host = host self.host = host
self.port = port self.port = port
self._url = url self._url = url
@ -221,14 +224,14 @@ class Application(web.Application):
def __get__(self, key): def __get__(self, key):
if self._rpc: if self._rpc:
return getattr(self._rpc,key) return getattr(self._rpc, key)
return getattr(self.arpc,key) return getattr(self.arpc, key)
@property @property
def url(self): def url(self):
if self._url: if self._url:
return self._url return self._url
return "http://{}:{}".format(self.host,self.port) return f"http://{self.host}:{self.port}"
@property @property
def rpc_url(self): def rpc_url(self):