123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109 |
- from pyexpat import model
- 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)
- 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
- 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"))
- sess.add(Item)
- 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):
- return json.dumps(object, cls=ComplexEncoder)
|