Skip to content

db.neo4j

Module contains class and utility functions to manage interactions with Neo4J database.

Neo4J

Class for managing interaction with Neo4J database.

Attributes:

Name Type Description
neo4j_url str

URL for connecting to Neo4j
neo4j_username str

Username for Neo4j database
neo4j_password str

Password for Neo4j database
use_env bool

True if above info are to be collected from env file.
driver GraphDatabase.driver

Neo4J driver instance
Source code in erad\db\neo4j_.py 
 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
154
155
156
157
158
159
160
161
162
class Neo4J:
    """Class for managing interaction with Neo4J database.

    Attributes:
        neo4j_url (str): URL for connecting to Neo4j
        neo4j_username (str): Username for Neo4j database
        neo4j_password (str): Password for Neo4j database
        use_env (bool): True if above info are to be collected
            from env file.
        driver (GraphDatabase.driver): Neo4J driver instance
    """

    def __init__(
        self,
        neo4j_url: Union[str, None] = None,
        neo4j_username: Union[str, None] = None,
        neo4j_password: Union[str, None] = None,
    ) -> None:
        """Constructor for Neo4J class.

        Args:
            neo4j_url (str): URL for connecting to Neo4j
            neo4j_username (str): Username for Neo4j database
            neo4j_password (str): Password for Neo4j database
        """

        self.neo4j_url = NEO4J_URL if NEO4J_URL else neo4j_url
        self.neo4j_username = (
            NEO4J_USERNAME if NEO4J_USERNAME else neo4j_username
        )
        self.neo4j_password = (
            NEO4J_PASSWORD if NEO4J_PASSWORD else neo4j_password
        )

        connection = Neo4jConnectionModel(
            neo4j_url=self.neo4j_url,
            neo4j_username=self.neo4j_username,
            neo4j_password=self.neo4j_password,
        )

        self.driver = GraphDatabase.driver(
            connection.neo4j_url,
            auth=basic_auth(
                connection.neo4j_username, connection.neo4j_password
            ),
        )

        logger.debug(
            f"Connected to {connection.neo4j_url} database successfully"
        )

    @staticmethod
    def rename_labels(label):
        """Method to replace the invalid character."""
        invalid_chars = ["-", ":", "(", ")", "."]
        for invalid_char in invalid_chars:
            if invalid_char in label:
                label = label.replace(invalid_char, "__")
        return label

    # def add_node(
    #     self,
    #     labels: Union[List, None] = None,
    #     properties: Union[Dict, None] = None,
    # ) -> None:
    #     """Method to add node to the Neo4j database.

    #     Args:
    #         labels (Union[List, None]): List of labels to be used for node
    #         properties (Union[Dict, None]): Properties to be used for the node
    #     """
    #     if labels is None:
    #         labels = []

    #     if properties is None:
    #         properties = {}

    #     labels = ":".join([self.rename_labels(label) for label in labels])
    #     cypher_query = "CREATE (:" + labels + " $properties)"

    #     with self.driver.session() as session:
    #         session.write_transaction(
    #             lambda tx: tx.run(cypher_query, properties=properties)
    #         )

    # def add_relationship(
    #     self,
    #     from_node_label: str,
    #     to_node_label: str,
    #     from_node: str,
    #     to_node: str,
    #     relationship_label: str,
    #     relationship_properties: Union[Dict, None] = None,
    # ) -> None:
    #     """Method to create relationship in graph database.

    #     Args:
    #         from_node_label (str): Node label for from node
    #         to_node_label (str): Node label for to node
    #         from_node (str): From node name
    #         to_node (str): To node name
    #         relationship_label (str): Relationship label name
    #         relationship_properties (Union[Dict, None]): Properties to be used
    #             for relationship
    #     """

    #     if relationship_properties is None:
    #         relationship_properties = {}

    #     cypher_query = (
    #         f"MATCH (a:{from_node_label}),(b:{to_node_label}) WHERE a.name = '{from_node}'"
    #         + f" AND b.name = '{to_node}' CREATE (a)-[:"
    #         + self.rename_labels(relationship_label)
    #         + " $properties]->(b)"
    #     )

    #     with self.driver.session() as session:
    #         session.write_transaction(
    #             lambda tx: tx.run(
    #                 cypher_query, properties=relationship_properties
    #             )
    #         )

    # def read_query(self, cypher_query: str) -> List:
    #     """Executes a Cypher read query."""

    #     with self.driver.session() as session:
    #         result = session.read_transaction(
    #             lambda tx: tx.run(cypher_query).data()
    #         )

    #     return result

    def close_driver(self):
        """Method to close the driver."""
        self.driver.close()

__init__(neo4j_url=None, neo4j_username=None, neo4j_password=None)

Constructor for Neo4J class.

Parameters:

Name Type Description Default
neo4j_url str

URL for connecting to Neo4j

 None
neo4j_username str

Username for Neo4j database

 None
neo4j_password str

Password for Neo4j database

 None
Source code in erad\db\neo4j_.py 
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
def __init__(
    self,
    neo4j_url: Union[str, None] = None,
    neo4j_username: Union[str, None] = None,
    neo4j_password: Union[str, None] = None,
) -> None:
    """Constructor for Neo4J class.

    Args:
        neo4j_url (str): URL for connecting to Neo4j
        neo4j_username (str): Username for Neo4j database
        neo4j_password (str): Password for Neo4j database
    """

    self.neo4j_url = NEO4J_URL if NEO4J_URL else neo4j_url
    self.neo4j_username = (
        NEO4J_USERNAME if NEO4J_USERNAME else neo4j_username
    )
    self.neo4j_password = (
        NEO4J_PASSWORD if NEO4J_PASSWORD else neo4j_password
    )

    connection = Neo4jConnectionModel(
        neo4j_url=self.neo4j_url,
        neo4j_username=self.neo4j_username,
        neo4j_password=self.neo4j_password,
    )

    self.driver = GraphDatabase.driver(
        connection.neo4j_url,
        auth=basic_auth(
            connection.neo4j_username, connection.neo4j_password
        ),
    )

    logger.debug(
        f"Connected to {connection.neo4j_url} database successfully"
    )

close_driver()

Method to close the driver.

Source code in erad\db\neo4j_.py 
160
161
162
def close_driver(self):
    """Method to close the driver."""
    self.driver.close()

rename_labels(label) staticmethod

Method to replace the invalid character.

Source code in erad\db\neo4j_.py 
78
79
80
81
82
83
84
85
@staticmethod
def rename_labels(label):
    """Method to replace the invalid character."""
    invalid_chars = ["-", ":", "(", ")", "."]
    for invalid_char in invalid_chars:
        if invalid_char in label:
            label = label.replace(invalid_char, "__")
    return label