openisp/Backend/Sources/Model/model_manager.py

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