Skip to content

CosmoTech_Acceleration_Library.Modelops.core.utils.model_util

ModelUtil

Utility class for Redis management

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class ModelUtil:
    """
    Utility class for Redis management
    """

    # ADT variables
    source_key = 'source'
    target_key = 'target'
    id_key = 'id'

    # Redis/Csm variables
    src_key = 'src'
    dest_key = 'dest'
    dt_id_key = 'id'

    @staticmethod
    def dict_to_cypher_parameters(parameters: dict) -> str:
        """
        Convert a dict to usable Cypher parameters object
        :param parameters: parameters dict
        :return: string representing parameters as Cyper Parameters
        """

        cypher_list = []
        for key, value in parameters.items():
            formatted_value = stringify_param_value(value)
            if isinstance(value, str):
                try:
                    json.loads(value)
                    formatted_value = json.dumps(value)
                except ValueError:
                    logger.debug(f"{value} is not a jsonString, use the raw value")
            cypher_list.append(f"{key} : {formatted_value}")
        joined_list = ', '.join(cypher_list)
        return '{' + joined_list + '}'

    @staticmethod
    def create_index_query(entity_name: str, entity_property_name: str) -> str:
        """
        Create an index query
        :param entity_name: the entity name on which you want to define an index
        :param entity_property_name:  the entity property name on which you want to define an index
        :return: the create index query
        """
        return f"CREATE INDEX ON :{entity_name}({entity_property_name})"

    @staticmethod
    def create_twin_query(twin_type: str, properties: dict) -> str:
        """
        Create a twin query
        :param twin_type:the future twin name
        :param properties: the properties of the twin
        :return: the create twin query
        """
        if ModelUtil.dt_id_key in properties:
            cypher_params = ModelUtil.dict_to_cypher_parameters(properties)
            return f"CREATE (:{twin_type} {cypher_params})"
        raise Exception(f"When you create a twin, you should define at least {ModelUtil.dt_id_key} properties ")

    @staticmethod
    def create_relationship_query(relationship_type: str, properties: dict) -> str:
        """
        Create a relationship query
        :param relationship_type: the future relationship name
        :param properties: the properties of the relationship (should contain 'src' and 'dest' properties)
        :return: the create relationship query
        """

        if ModelUtil.src_key in properties and ModelUtil.dest_key in properties:
            cypher_params = ModelUtil.dict_to_cypher_parameters(properties)
            return f"MATCH (n), (m) WHERE n.{ModelUtil.dt_id_key} = '{properties.get(ModelUtil.src_key)}' " \
                   f"AND m.{ModelUtil.dt_id_key} = '{properties.get(ModelUtil.dest_key)}' " \
                   f"CREATE (n)-[r:{relationship_type} {cypher_params}]->(m) RETURN r"
        raise Exception(
            f"When you create a relationship, you should define at least {ModelUtil.src_key} and {ModelUtil.dest_key} properties "
        )

    @staticmethod
    def dict_to_json(obj: dict) -> str:
        """
        Transform a dict to a json string
        :param obj: the dict
        :return: the json string corresponding
        """
        return json.dumps(obj, indent=2)

    @staticmethod
    def result_set_to_json(query_result: QueryResult) -> list:
        """
        Transform a QueryResult object to a json string list
        :param query_result: the QueryResult object
        :return: the json string list
        """
        flattened_headers = [item for sublist in query_result.header for item in sublist]
        headers_without_integers = [x for x in flattened_headers if not isinstance(x, int)]
        result_list = []
        for result in query_result.result_set:
            result_dict = {}
            for i in range(len(headers_without_integers)):
                obj = result[i]
                if isinstance(obj, Edge) or isinstance(obj, Node):
                    result_dict[headers_without_integers[i]] = obj.properties
                else:
                    result_dict[headers_without_integers[i]] = obj
            result_list.append(ModelUtil.dict_to_json(result_dict))
        return result_list

    @staticmethod
    def print_query_result(query_result: QueryResult) -> None:
        """
        Pretty print a QueryResult
        :param query_result: the QueryResult to print
        """
        list_to_print = ModelUtil.result_set_to_json(query_result)
        for result in list_to_print:
            print(result)

    @staticmethod
    def convert_datetime_to_str(date: datetime) -> str:
        """
        Convert a datetime to a str
        :param date: the datetime
        :return: the string representing the datetime
        """
        return date.strftime('%Y/%m/%d - %H:%M:%S')

    @staticmethod
    def convert_str_to_datetime(date_str: str) -> datetime:
        """
        Convert a datetime to a str
        :param date_str: the str representing a date
        :return: the datetime corresponding to date_str
        """
        date_time_obj = datetime.strptime(date_str, '%Y/%m/%d - %H:%M:%S')
        return date_time_obj

    @staticmethod
    def build_graph_key_pattern(graph_name: str) -> str:
        return graph_name + ":*"

convert_datetime_to_str(date) staticmethod

Convert a datetime to a str :param date: the datetime :return: the string representing the datetime

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
132
133
134
135
136
137
138
139
@staticmethod
def convert_datetime_to_str(date: datetime) -> str:
    """
    Convert a datetime to a str
    :param date: the datetime
    :return: the string representing the datetime
    """
    return date.strftime('%Y/%m/%d - %H:%M:%S')

convert_str_to_datetime(date_str) staticmethod

Convert a datetime to a str :param date_str: the str representing a date :return: the datetime corresponding to date_str

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
141
142
143
144
145
146
147
148
149
@staticmethod
def convert_str_to_datetime(date_str: str) -> datetime:
    """
    Convert a datetime to a str
    :param date_str: the str representing a date
    :return: the datetime corresponding to date_str
    """
    date_time_obj = datetime.strptime(date_str, '%Y/%m/%d - %H:%M:%S')
    return date_time_obj

create_index_query(entity_name, entity_property_name) staticmethod

Create an index query :param entity_name: the entity name on which you want to define an index :param entity_property_name: the entity property name on which you want to define an index :return: the create index query

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
51
52
53
54
55
56
57
58
59
@staticmethod
def create_index_query(entity_name: str, entity_property_name: str) -> str:
    """
    Create an index query
    :param entity_name: the entity name on which you want to define an index
    :param entity_property_name:  the entity property name on which you want to define an index
    :return: the create index query
    """
    return f"CREATE INDEX ON :{entity_name}({entity_property_name})"

create_relationship_query(relationship_type, properties) staticmethod

Create a relationship query :param relationship_type: the future relationship name :param properties: the properties of the relationship (should contain 'src' and 'dest' properties) :return: the create relationship query

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@staticmethod
def create_relationship_query(relationship_type: str, properties: dict) -> str:
    """
    Create a relationship query
    :param relationship_type: the future relationship name
    :param properties: the properties of the relationship (should contain 'src' and 'dest' properties)
    :return: the create relationship query
    """

    if ModelUtil.src_key in properties and ModelUtil.dest_key in properties:
        cypher_params = ModelUtil.dict_to_cypher_parameters(properties)
        return f"MATCH (n), (m) WHERE n.{ModelUtil.dt_id_key} = '{properties.get(ModelUtil.src_key)}' " \
               f"AND m.{ModelUtil.dt_id_key} = '{properties.get(ModelUtil.dest_key)}' " \
               f"CREATE (n)-[r:{relationship_type} {cypher_params}]->(m) RETURN r"
    raise Exception(
        f"When you create a relationship, you should define at least {ModelUtil.src_key} and {ModelUtil.dest_key} properties "
    )

create_twin_query(twin_type, properties) staticmethod

Create a twin query :param twin_type:the future twin name :param properties: the properties of the twin :return: the create twin query

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
61
62
63
64
65
66
67
68
69
70
71
72
@staticmethod
def create_twin_query(twin_type: str, properties: dict) -> str:
    """
    Create a twin query
    :param twin_type:the future twin name
    :param properties: the properties of the twin
    :return: the create twin query
    """
    if ModelUtil.dt_id_key in properties:
        cypher_params = ModelUtil.dict_to_cypher_parameters(properties)
        return f"CREATE (:{twin_type} {cypher_params})"
    raise Exception(f"When you create a twin, you should define at least {ModelUtil.dt_id_key} properties ")

dict_to_cypher_parameters(parameters) staticmethod

Convert a dict to usable Cypher parameters object :param parameters: parameters dict :return: string representing parameters as Cyper Parameters

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@staticmethod
def dict_to_cypher_parameters(parameters: dict) -> str:
    """
    Convert a dict to usable Cypher parameters object
    :param parameters: parameters dict
    :return: string representing parameters as Cyper Parameters
    """

    cypher_list = []
    for key, value in parameters.items():
        formatted_value = stringify_param_value(value)
        if isinstance(value, str):
            try:
                json.loads(value)
                formatted_value = json.dumps(value)
            except ValueError:
                logger.debug(f"{value} is not a jsonString, use the raw value")
        cypher_list.append(f"{key} : {formatted_value}")
    joined_list = ', '.join(cypher_list)
    return '{' + joined_list + '}'

dict_to_json(obj) staticmethod

Transform a dict to a json string :param obj: the dict :return: the json string corresponding

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
92
93
94
95
96
97
98
99
@staticmethod
def dict_to_json(obj: dict) -> str:
    """
    Transform a dict to a json string
    :param obj: the dict
    :return: the json string corresponding
    """
    return json.dumps(obj, indent=2)

print_query_result(query_result) staticmethod

Pretty print a QueryResult :param query_result: the QueryResult to print

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
122
123
124
125
126
127
128
129
130
@staticmethod
def print_query_result(query_result: QueryResult) -> None:
    """
    Pretty print a QueryResult
    :param query_result: the QueryResult to print
    """
    list_to_print = ModelUtil.result_set_to_json(query_result)
    for result in list_to_print:
        print(result)

result_set_to_json(query_result) staticmethod

Transform a QueryResult object to a json string list :param query_result: the QueryResult object :return: the json string list

Source code in CosmoTech_Acceleration_Library/Modelops/core/utils/model_util.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@staticmethod
def result_set_to_json(query_result: QueryResult) -> list:
    """
    Transform a QueryResult object to a json string list
    :param query_result: the QueryResult object
    :return: the json string list
    """
    flattened_headers = [item for sublist in query_result.header for item in sublist]
    headers_without_integers = [x for x in flattened_headers if not isinstance(x, int)]
    result_list = []
    for result in query_result.result_set:
        result_dict = {}
        for i in range(len(headers_without_integers)):
            obj = result[i]
            if isinstance(obj, Edge) or isinstance(obj, Node):
                result_dict[headers_without_integers[i]] = obj.properties
            else:
                result_dict[headers_without_integers[i]] = obj
        result_list.append(ModelUtil.dict_to_json(result_dict))
    return result_list