Skip to content

CosmoTech_Acceleration_Library.Modelops.core.io.model_reader

ModelReader

Bases: GraphHandler

Model Reader for cached data

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
 12
 13
 14
 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
class ModelReader(GraphHandler):
    """
    Model Reader for cached data
    """

    def get_twin_types(self) -> list:
        """
        Get twin types
        :return: twin types list
        """
        return [item for sublist in self.graph.labels() for item in sublist]

    def get_twins_by_type(self, twin_type: str, limit: int = 0) -> QueryResult:
        """
        Get twins by type
        :param twin_type: the twin type requested
        :param limit: the limit number of twin retrieved
        :return: the twin list corresponding to twin type parameter
        """
        twin_query = f'MATCH (node:{twin_type}) RETURN node'
        if limit != 0:
            twin_query = f'{twin_query} LIMIT {str(limit)}'
        logger.debug(f"Query : {twin_query}")
        return self.graph.query(twin_query, read_only=True)

    def get_twin_properties_by_type(self, twin_type: str) -> list:
        """
        Get twin properties regarding a twin_type
        Note: this will work if all twin (with the same type) have same properties set
        :param twin_type: the twin type
        :return: the properties list
        """
        result = []
        twin_result = self.get_twins_by_type(twin_type, 1)
        result_set = twin_result.result_set
        if result_set and result_set[0]:
            for key, val in result_set[0][0].properties.items():
                if str(key) != ModelUtil.dt_id_key:
                    result.append(str(key))
                else:
                    result.append(ModelUtil.id_key)
        return result

    def get_relationship_types(self) -> list:
        """
        Get relationship types
        :return: relationship types list
        """
        return [item for sublist in self.graph.relationship_types() for item in sublist]

    def get_relationships_by_type(self, relationship_type: str, limit: int = 0) -> QueryResult:
        """
        Get relationships by type
        :param relationship_type: the relationship type requested
        :param limit: the limit number of twin retrieved
        :return: the relationship list corresponding to relationship type parameter
        """
        rel_query = f'MATCH (n)-[relation:{relationship_type}]->(m) RETURN n.{ModelUtil.dt_id_key} as {ModelUtil.source_key}, ' \
                    f'm.{ModelUtil.dt_id_key} as {ModelUtil.target_key}, relation'
        if limit != 0:
            rel_query = f'{rel_query} LIMIT {str(limit)}'
        logger.debug(f"Query : {rel_query}")
        return self.graph.query(rel_query, read_only=True)

    def get_relationship_properties_by_type(self, relationship_type: str) -> list:
        """
        Get relationship properties regarding a relationship_type
        Note: this will work if all relationship (with the same type) have same properties set
        :param relationship_type: the relationship type
        :return: the properties list
        """
        result = [ModelUtil.source_key, ModelUtil.target_key]
        relationship_result = self.get_relationships_by_type(relationship_type, 1)
        result_set = relationship_result.result_set
        if result_set and result_set[0]:
            # relationship
            for key, val in result_set[0][2].properties.items():
                if not str(key) in result:
                    if str(key) == ModelUtil.dt_id_key:
                        result.append(ModelUtil.id_key)
                    elif str(key) != ModelUtil.src_key and str(key) != ModelUtil.dest_key:
                        result.append(str(key))
        return result

    def query(self, query: str, params: dict = None, timeout: int = None, read_only: bool = False) -> QueryResult:
        """
        Run specified query
        :param query: the query to run
        :param params: the parameters for the query if any
        :param timeout: a specific timeout
        :param read_only: executes a readonly query if set to True
        :return: the QueryResult corresponding to specified query
        """
        logger.debug(f"Query : {query} with params : {params}")
        return self.graph.query(q=query, params=params, timeout=timeout, read_only=read_only)

    def exists(self, key) -> bool:
        """
        Check if a key exists in Redis
        :param key: the key
        :return: True if exists else False
        """
        return False if self.r.exists(key) == 0 else True

exists(key)

Check if a key exists in Redis :param key: the key :return: True if exists else False

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
108
109
110
111
112
113
114
def exists(self, key) -> bool:
    """
    Check if a key exists in Redis
    :param key: the key
    :return: True if exists else False
    """
    return False if self.r.exists(key) == 0 else True

get_relationship_properties_by_type(relationship_type)

Get relationship properties regarding a relationship_type Note: this will work if all relationship (with the same type) have same properties set :param relationship_type: the relationship type :return: the properties list

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def get_relationship_properties_by_type(self, relationship_type: str) -> list:
    """
    Get relationship properties regarding a relationship_type
    Note: this will work if all relationship (with the same type) have same properties set
    :param relationship_type: the relationship type
    :return: the properties list
    """
    result = [ModelUtil.source_key, ModelUtil.target_key]
    relationship_result = self.get_relationships_by_type(relationship_type, 1)
    result_set = relationship_result.result_set
    if result_set and result_set[0]:
        # relationship
        for key, val in result_set[0][2].properties.items():
            if not str(key) in result:
                if str(key) == ModelUtil.dt_id_key:
                    result.append(ModelUtil.id_key)
                elif str(key) != ModelUtil.src_key and str(key) != ModelUtil.dest_key:
                    result.append(str(key))
    return result

get_relationship_types()

Get relationship types :return: relationship types list

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
55
56
57
58
59
60
def get_relationship_types(self) -> list:
    """
    Get relationship types
    :return: relationship types list
    """
    return [item for sublist in self.graph.relationship_types() for item in sublist]

get_relationships_by_type(relationship_type, limit=0)

Get relationships by type :param relationship_type: the relationship type requested :param limit: the limit number of twin retrieved :return: the relationship list corresponding to relationship type parameter

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
def get_relationships_by_type(self, relationship_type: str, limit: int = 0) -> QueryResult:
    """
    Get relationships by type
    :param relationship_type: the relationship type requested
    :param limit: the limit number of twin retrieved
    :return: the relationship list corresponding to relationship type parameter
    """
    rel_query = f'MATCH (n)-[relation:{relationship_type}]->(m) RETURN n.{ModelUtil.dt_id_key} as {ModelUtil.source_key}, ' \
                f'm.{ModelUtil.dt_id_key} as {ModelUtil.target_key}, relation'
    if limit != 0:
        rel_query = f'{rel_query} LIMIT {str(limit)}'
    logger.debug(f"Query : {rel_query}")
    return self.graph.query(rel_query, read_only=True)

get_twin_properties_by_type(twin_type)

Get twin properties regarding a twin_type Note: this will work if all twin (with the same type) have same properties set :param twin_type: the twin type :return: the properties list

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def get_twin_properties_by_type(self, twin_type: str) -> list:
    """
    Get twin properties regarding a twin_type
    Note: this will work if all twin (with the same type) have same properties set
    :param twin_type: the twin type
    :return: the properties list
    """
    result = []
    twin_result = self.get_twins_by_type(twin_type, 1)
    result_set = twin_result.result_set
    if result_set and result_set[0]:
        for key, val in result_set[0][0].properties.items():
            if str(key) != ModelUtil.dt_id_key:
                result.append(str(key))
            else:
                result.append(ModelUtil.id_key)
    return result

get_twin_types()

Get twin types :return: twin types list

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
17
18
19
20
21
22
def get_twin_types(self) -> list:
    """
    Get twin types
    :return: twin types list
    """
    return [item for sublist in self.graph.labels() for item in sublist]

get_twins_by_type(twin_type, limit=0)

Get twins by type :param twin_type: the twin type requested :param limit: the limit number of twin retrieved :return: the twin list corresponding to twin type parameter

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
24
25
26
27
28
29
30
31
32
33
34
35
def get_twins_by_type(self, twin_type: str, limit: int = 0) -> QueryResult:
    """
    Get twins by type
    :param twin_type: the twin type requested
    :param limit: the limit number of twin retrieved
    :return: the twin list corresponding to twin type parameter
    """
    twin_query = f'MATCH (node:{twin_type}) RETURN node'
    if limit != 0:
        twin_query = f'{twin_query} LIMIT {str(limit)}'
    logger.debug(f"Query : {twin_query}")
    return self.graph.query(twin_query, read_only=True)

query(query, params=None, timeout=None, read_only=False)

Run specified query :param query: the query to run :param params: the parameters for the query if any :param timeout: a specific timeout :param read_only: executes a readonly query if set to True :return: the QueryResult corresponding to specified query

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_reader.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
def query(self, query: str, params: dict = None, timeout: int = None, read_only: bool = False) -> QueryResult:
    """
    Run specified query
    :param query: the query to run
    :param params: the parameters for the query if any
    :param timeout: a specific timeout
    :param read_only: executes a readonly query if set to True
    :return: the QueryResult corresponding to specified query
    """
    logger.debug(f"Query : {query} with params : {params}")
    return self.graph.query(q=query, params=params, timeout=timeout, read_only=read_only)