from pyexpat import model from sys import modules from tokenize import String from sqlalchemy.ext.declarative import DeclarativeMeta from sqlalchemy import null from utility.app_logging import * import json from werkzeug.security import generate_password_hash from utility.app_logging import logger_name import logging logger = logging.getLogger(logger_name + ".MODEL") def declare_model(): # insert here all model that need to be declared # the fact import models files # OpenISP Core Model import Model.isp_model # OpenISP Inventory Model import Modules.Inventory.inventory_model def init() : import Model.isp_model import persistence import Modules.Inventory.inventory_model with persistence.get_Session_Instance() as sess : ##TODO maybe not the best place to perform that Item = sess.query(Model.isp_model.user_account).filter(Model.isp_model.user_account.nickname == "admin").first() if not isinstance(Item,Model.isp_model.user_account): Item = Model.isp_model.user_account(id = 1, nickname = "admin",description = "admin account", is_super_admin = True,password = generate_password_hash("aseqzdwxc")) obj = Modules.Inventory.inventory_model.inventory_item(name = "mikrotik rb2011 saline 1",brand= "mikrotik",note="router saline2") sess.add(Item) sess.add(obj) sess.commit() logger.debug("Admin account created") else : logger.debug("Admin account already created") # from https://stackoverflow.com/questions/5022066/how-to-serialize-sqlalchemy-result-to-json class ComplexEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj.__class__, DeclarativeMeta): # an SQLAlchemy class fields = {} for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']: data = obj.__getattribute__(field) try: # this will fail on non-encodable values, like other classes json.dumps(data) fields[field] = data except TypeError: fields[field] = None # a json-encodable dict return fields return json.JSONEncoder.default(self, obj) # from https://howtodoinjava.com/json/custom-deserialization/ class ComplexDecoder(json.JSONDecoder): def __init__(self): json.JSONDecoder.__init__( self, object_hook=self.dict_to_object, ) def dict_to_object(self, d): if '__class__' in d: class_name = d.pop('__class__') module_name = d.pop('__module__') module = __import__(module_name) class_ = getattr(module, class_name) args = { key: value for key, value in d.items() } inst = class_(**args) else: inst = d return inst def setModelItemAttributesFromJson(Item, JsonString): attributes_dict = null try: attributes_dict = json.loads( JsonString, cls=ComplexDecoder) except: raise "input is not a Json String" # verification of attribute """ for key in attributes_dict: print(key) print (Item.__dict__) if not key in Item.__dict__ : raise "one input attribute is not an attribute of the destination object" """ # TODO later verify json schema ? for key in attributes_dict: setattr(Item, key, attributes_dict[key]) def ModelObjectToJsonString(object) -> str: json_string = json.dumps(object, cls=ComplexEncoder) json_dict : dict = json.loads(json_string) json_dict.pop("registry") # removing the registry item that comes from the sql alchemy type. json_string = json.dumps(json_dict,indent=4) assert isinstance(json_string,str) return json_string