PK!crccvremoteapi/__init__.pyfrom remote_api import RemoteApi, Interface, create_remote_json_call __version__ = '0.1.0' PK!=ۆD D vremoteapi/main.py from typing import Optional, Dict, Any, Union, List, Callable from typing import NamedTuple import json # a list of functions that may be called Interface = List[ Callable[[Any], Any] ] Params = Optional[ Union[Dict[str, Any], List[Any] ] ] # representa o objeto de chamada remota class FuncCall(NamedTuple): ''' params = may be xargs=dict or args=list format ''' funcname: str params: Params def _json_to_dict(_json): return json.loads(_json) def _dict_to_json(d: Dict[str, Any]): return json.dumps(d) def _call_no_args(_func:str) -> Any: return _func() # used when params are a list=*args def _call_args(_func, args: List[Any]) -> Any: return _func(*args) # used when params are a dict=*xargs def _call_xargs(_func, xargs: Dict[str, Any]) -> Any: return _func(**xargs) # interface is a list with remote callable functions # NOTE: This function DO NOT cast the function result def run(_json, api: Interface) -> Any: # load json d = _json_to_dict(_json) fname = d['funcname'] _params = d['params'] # select choosen functoin from function interface _func = None for func in api: if func.__name__ == fname: _func = func break if _func is None: raise ValueError("Tentativa de chamada de funcao remota falhou. Funcao não localizada na api") # call function if _params == [] or _params is None: res = _call_no_args(_func) elif isinstance(_params, list): res = _call_args(_func, _params) elif isinstance(_params, dict): res = _call_xargs(_func, _params) else: raise ValueError(f"Tentativa de chamada de funcao remota {funcname} falhou. Parametro(s) da funcao precisa ser ou vazio, ou uma lista, ou um dicionario") # create response return res # as dict def cast_result(data: Any) -> Dict[str, Any]: res = { 'result' : data } return res # helper for log/debug def create_remote_json_call(funcname:str, params: Params) -> str: return _dict_to_json(FuncCall(funcname, params)._asdict()) # -------------- TEST def me(): print("me running...") return ['com1', 'com2', None] def metoo(a, b): print("metoo running...") return {'a=': a, 'b=' : b, 'c=': ['com1', 'com2'], 'nada': None} if __name__ == '__main__': api: Interface = [me, metoo] # function without params _json = create_remote_json_call('me', None) print(_json) print(_json_to_dict(_json)) res = run(_json, api) res = cast_result(res) print(_dict_to_json(res)) # function with list arguments _json = create_remote_json_call('metoo', [10,20]) print(_json) print(_json_to_dict(_json)) res = run(_json, api) res = cast_result(res) print(_dict_to_json(res)) # function with xargs arguments _json = create_remote_json_call('metoo', {'a':66, 'b': 77}) print(_json) print(_json_to_dict(_json)) res = run(_json, api) res = cast_result(res) print(_dict_to_json(res))PK!vremoteapi/OLD_/__init__.pyPK!xD;vremoteapi/OLD_/command.pyfrom typing import NamedTuple, List, Any, Optional # a ideia é: passa um commando com parametros, e retorna um tuple com resultado ou erro class Value(NamedTuple): type_: str # int, str! (maybe on future) -> float, binary, bool... content: Any class Parameter(NamedTuple): par_name: str value: Value # ATTENTION: In this version Dictionary is always Flat !! Dictionary = List[Parameter] # representa o objeto de commando class Command(NamedTuple): cmd_uuid: str parameters: Optional[Dictionary] = None class ErrorInfo(NamedTuple): message: bool has_error: bool class Answer(NamedTuple): response: Optional[Dictionary] = None error: Optional[ErrorInfo] = None history: Optional[List[Any]] = None if __name__ == '__main__': a = Answer() import json print(json.dumps(a._asdict())) passPK!*RRvremoteapi/OLD_/dispatcher.pyfrom typing import Callable # Recebe stream de commandos em json, verifica integridade e assinatura, e caso ok dispacha para funcao de modo # sincrono. Ao receber a resposta da funcao, reempacota resposta em json. ''' def _jsonToCommand(json) -> Command: raise TypeError("Wrong signature!") pass def _bind_run(c:Command, func) -> Answer: pars: Dictionary = _extract_parameters_from_command(Command) try: answer: = run(pars) except: # an exception was raised at-run-time during function execution ans: Answer = _convert_function_return_to_Answer(resp) return ans Answer_Signature = DictSignature Param_Signature = DictSignature Func_Name_UUID = str Function = Callable[[Any], Answer] FunctionSignature = Tuple[Func_Name_UUID, Function, Param_Signature, Answer_Signature] # ex: func = ('listaCom', None, 'list') def addFunction(fun: VFunction): pass ''' def me(): print("me running...") pass def metoo(): print("metoo running...") pass if __name__ == '__main__': from command import Command from typing import NamedTuple import json inidata = { 'cmd_uuid': 'serial_ports', 'parameters': {'a':1, 'b':2, 'd': 'juca', 'c': None} } cmd = Command(**inidata) print(cmd) parsed = json.dumps(cmd._asdict()) print(parsed) d = json.loads(parsed) print(inidata == d) cmd = Command(**d) print(cmd.parameters) print(json.dumps(d)) li = [me, metoo] for f in li: callf = f.__name__ + '()' eval(callf) PK![ vremoteapi/OLD_/factories.py#helpers factories and conversors from typing import Dict, Tuple, Optional, Any, List from command import Dictionary, Parameter, Value from types_ import Type from type_checking import isTypeCorrect ErrMsg = Optional[str] def create_Value_from_int(i:int) -> Value: return Value(Type.int,i) def create_Value_from_str(s:str) -> Value: return Value(Type.str,s) def create_Value_from_bytes(b:bytes) -> Value: l: List = [] for each in b: l.append(each) return Value(Type.bytes,l) def create_Value_from_list(li:list) -> Value: l: List = [] for each in li: l.append(each) return Value(Type.list, l) def create_Value_from_None() -> Value: return Value(Type.none,None) def create_Value_From_Any(d: Any) -> Tuple[Value, bool, ErrMsg]: if isinstance(d, str): return create_Value_from_str(d), False, None elif isinstance(d, int): return create_Value_from_int(d), False, None elif isinstance(d, bytes): # NOTE: there is no difference currently between list and bytes. bytes is a list return create_Value_from_bytes(d), False, None elif isinstance(d, list): return create_Value_from_list(d), False, None elif d is None: return create_Value_from_None(), False, None else: return None, True, 'type not suported' def flatdict_to_flatDictionary(dic: Dict) -> Optional[Dictionary]: if dic == []: return None Dic: Dictionary = [] for key,value in dic.items(): value_, isErr, errmsg = create_Value_From_Any(value) if isErr: err = f"Can't create Dictionary from dict specified. Key '{key}' of the dict has a value '{value}' unsupported." raise ValueError(err) param = Parameter(key, value_) Dic.append(param) return Dic def _convertValue_to_pythondata(v: Value): if v.type_ == Type().int: return v.content elif v.type_ == Type().str: return v.content elif v.type_ == Type().bytes: return bytes(v.content) # v.content is a list of integer each one representing a byte elif v.type_ == Type().list: return v.content # v.content is a list of integer each one representing a byte elif v.type_ == Type().none: return None # v.content is a list of integer each one representing a byte else: raise TypeError(f"Cannot convert py-Type {v} to Value object. Conversion routine not implemented.") def flatDictionary_to_flatdict(Dic: Dictionary) -> Optional[Dict]: if Dic is None: return None dic = {} for param in Dic: p: Parameter = param key = p.par_name (isCorrect, errMsg) = isTypeCorrect(p.value) if not isCorrect: raise TypeError(errMsg) value = _convertValue_to_pythondata(p.value) dic[key] = value return dic if __name__ == '__main__': dic = {'xip': 2, 'msg': 'elemtar caro whatson', 'stream': b'Flavio', 'nomes': ['fla', 'flu'], 'nenhum': None} D: Dictionary = flatdict_to_flatDictionary(dic) print(D) dic = flatDictionary_to_flatdict(D) print(dic) passPK!S $vremoteapi/OLD_/require_signature.py# permite checar assinatura de funcoes (se tipos correspondem a uma determinada assinatura) from command import Dictionary, Parameter from typing import List, Tuple, Optional from type_checking import isTypeCorrect Type = str ErrMsg = Optional[str] DictSignature = List[Tuple[str, Type]] # ex: [('nome', 'str'), ('idade','int'), ( 'packet', 'bytes' )] # checa apenas p tipo do valor def isFlatDictSignatureCorrect(dic: Dictionary, sig: DictSignature) -> Tuple[bool, ErrMsg]: expectedSig:DictSignature = sig realSig: DictSignature = [] # get real signature for param in dic: p: Parameter = param # confirme if param value is right (isOk, errMsg) = isTypeCorrect(p.value) if not isOk: return False, f"Assinatura de Dictionary incorreta. Parametro '{param}' possui tipo imcompativel com valor. Detalhe: {errMsg}" # get real param signature par_name = p.par_name type_ = p.value.type_ tuple_ = (par_name, type_) realSig.append( tuple_ ) # compare real w/ expected hasError = False if not len(expectedSig) == len(realSig): errmsg = f"Assinatura de Dictionary invalida. Assinatura esperada é '{expectedSig}', porem assinatura atual é {realSig}. " return False, errmsg for k in range(len(expectedSig)): if not realSig[k] == expectedSig[k]: errmsg = f"Assinatura de Dictionary invalida. Assinatura esperada é '{expectedSig}', porem assinatura atual é {realSig}. " errmsg = errmsg + f"Observe que o parametro '{realSig[k]}' é diferente do esperado '{expectedSig[k]}'" return False, errmsg return True, None if __name__ == '__main__': d = {'a':2, 'z':66, 'texto': 'oi malandro', 'data': b'Flavinho', 'ports': ['com1', 'com2'], 'nenhum': None} from factories import flatdict_to_flatDictionary as dict_to_Dictionary dic = dict_to_Dictionary(d) sig = [ ('a','int'), ('z','int'), ('texto','str'), ('data','bytes'), ('ports', 'list'), ('nenhum', 'none') ] print(d) print(dic) print(sig) res = isOk, errMsg = isFlatDictSignatureCorrect(dic, sig) print(res) passPK!ڲ55 vremoteapi/OLD_/type_checking.py# check if data type is right from command import Value from typing import Tuple, List, Optional from types_ import suported_types, Type # NOTA: o tipo de conteudo futuramente pode ser compativel com mais de um Type da libraria, por isto o retorno é uma lista def _getMyPossibleTypes(content) -> Optional[List[str]]: if isinstance(content, str): return [Type().str] if isinstance(content, int): return [Type().int] if isinstance(content, list): return [Type().bytes, Type().list] if content is None: return [Type().none] else: return None def _isSuportedType(type_) -> bool: for each in suported_types: if each == type_: return True return False ErrMsg = Optional[str] def isTypeCorrect(value: Value) -> Tuple[bool, ErrMsg]: # check if type informed is suported if not _isSuportedType(value.type_): return False, f"Tipo informado '{value.type_}' não é valido. Tipos validos são: {suported_types}" # check if type informed represent the value.content possibleRealDataTypes: List[str] = _getMyPossibleTypes(value.content) informedType = value.type_ match = False for p in possibleRealDataTypes: #print(p) if p== informedType: match = True if match == False: errmsg = f"Tipo do valor invalido. Tipo informado é '{informedType}', porém o tipo real detectado é algum destes {possibleRealDataTypes}." return False, errmsg else: return True, None if __name__ == '__main__': v = Value('list', [12,12,'com1','com2']) v = Value('none', None) tuple_ = (isCorrect, errMsg) = isTypeCorrect(v) dic = v._asdict() import json print(json.dumps(dic)) print(tuple_)PK!`Pvremoteapi/OLD_/types_.pyfrom typing import NamedTuple # tipos considerados validos # este objeto na verdade funciona como um Enum class Type(NamedTuple): str = 'str' int = 'int' bytes = 'bytes' #bytes sao armazenados como uma lista de inteiros (entre 0<255) list = 'list' #heterogeneous list none = 'none' # universo de tipos suportados suported_types = [ Type().str, Type().int, Type().bytes, Type().list, Type().none ] PK!vremoteapi/remote_api.py# helper object from main import Interface, run, cast_result, create_remote_json_call from typing import Dict, Any from typing import NamedTuple class RemoteApi: def __init__(self, api: Interface) -> None: self._api = api def _run_dont_cast_result(self, _json:str) -> Any: return run(_json, self._api) def run(self, _json:str) -> Dict[str, Any]: result = self._run_dont_cast_result(_json) return cast_result(result) # -------- Test def me(): print("me running...") pass def metoo(a, b): print("metoo running...") return {'a=': a, 'b=' : b} def me2(): print("me2 running...") pass def metoo2(a, b): print("metoo2 running...") return {'a=': a, 'b=' : b} if __name__ == '__main__': api: Interface = [ me, metoo, me2, metoo2 ] remoteApi = RemoteApi(api) rcall = create_remote_json_call calls = [ rcall('me', None), rcall('me2', None), rcall('metoo2', [1,2]), rcall('metoo2', [1, 'flavitcho']), rcall('metoo', [1, {'dict': 'olha só'}]), rcall('NON_EXISTENT_FUNC', [1, {'dict': 'olha só'}]) ] for _call in calls: res = remoteApi.run(_call) print(res) PK!HnHTU vremoteapi-0.1.1.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H.+ #vremoteapi-0.1.1.dist-info/METADATANJ1)еR!(kQMw5? }{"⵷C{ؼSLsݪ ̑$d‰m_ƽ Y$a(9ݡ\ -|"16L15Z=uZ%w%B40mHG#֤m.)5}7K: u?bJ<0m .{k`*~z:q~PK!HqV[q!vremoteapi-0.1.1.dist-info/RECORD}ˎP}? ts rS r#c2 {:EJ[XV=# sa^XE +w*ahuU{|4mk\OU0ƿXh׃cʎy|K͓+:H}뚝mIXR!𲲯r"QegRbU1&?;ed<8<6uCUɴe { PK!crccvremoteapi/__init__.pyPK!=ۆD D vremoteapi/main.pyPK! vremoteapi/OLD_/__init__.pyPK!xD;D vremoteapi/OLD_/command.pyPK!*RRvremoteapi/OLD_/dispatcher.pyPK![ vremoteapi/OLD_/factories.pyPK!S $r$vremoteapi/OLD_/require_signature.pyPK!ڲ55 y-vremoteapi/OLD_/type_checking.pyPK!`P4vremoteapi/OLD_/types_.pyPK!6vremoteapi/remote_api.pyPK!HnHTU <vremoteapi-0.1.1.dist-info/WHEELPK!H.+ #<vremoteapi-0.1.1.dist-info/METADATAPK!HqV[q!=vremoteapi-0.1.1.dist-info/RECORDPK s@