Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, user, password, url ='bolt://localhost:7687', debug = False):
self.debug = debug
self.neo4j_url = url
self.neo4j_user = user
self.neo4j_password = password
self.driver = neo4j.v1.GraphDatabase.driver(self.neo4j_url,
auth=(self.neo4j_user,
self.neo4j_password))
def __init__(self, uri, user, password):
self._driver = GraphDatabase.driver(uri, auth=(user, password), max_retry_time=1)
return driver
def create_demo_user(neo4j_session):
results = neo4j_session.run("MATCH (u:User {user : 'demo'}) RETURN u")
result = results.single()
if not result:
print("Adding user 'demo' to the neo4j database.")
neo4j_session.run("CREATE (a:User {user: {username}, argon_password: {hash}})",
{"username": "demo", "hash": argon2.hash("demo123")})
# For use in easy_install.sh to set up a demo user.
if __name__ == '__main__':
uri = sys.argv[1]
password = sys.argv[2]
driver = GraphDatabase.driver(uri, auth=basic_auth("neo4j", password))
with driver.session() as neo4j_session:
create_demo_user(neo4j_session)
"""
Execute the cartography.sync.Sync.run method with parameters built from the given configuration object.
This function will create a Neo4j driver object from the given Neo4j configuration options (URI, auth, etc.) and
will choose a sensible update tag if one is not specified in the given configuration.
:type sync: cartography.sync.Sync
:param sync: A sync task to run.
:type config: cartography.config.Config
:param config: The configuration to use to run the sync task.
"""
neo4j_auth = None
if config.neo4j_user or config.neo4j_password:
neo4j_auth = (config.neo4j_user, config.neo4j_password)
try:
neo4j_driver = GraphDatabase.driver(
config.neo4j_uri,
auth=neo4j_auth,
)
except neobolt.exceptions.ServiceUnavailable as e:
logger.debug("Error occurred during Neo4j connect.", exc_info=True)
logger.error(
(
"Unable to connect to Neo4j using the provided URI '%s', an error occurred: '%s'. Make sure the Neo4j "
"server is running and accessible from your network."
),
config.neo4j_uri,
e,
)
return
except neobolt.exceptions.AuthError as e:
logger.debug("Error occurred during Neo4j auth.", exc_info=True)
Relation type to use for edges encoding homomorphisms.
graph_relation_label : str, optional
Relation type to use for edges encoding relations.
"""
# The following idea is cool but it's not so easy:
# as we have two types of nodes in the hierarchy:
# graphs and rules, as well as two types of edges:
# homomorphisms and relations, and so far Neo4jGraph
# supports only a single label for nodes and for edges
# Neo4jGraph.__init__(
# self, uri=uri, user=user, password=password,
# node_label="hierarchyNode",
# edge_label="hierarchyEdge")
if driver is None:
self._driver = GraphDatabase.driver(
uri, auth=(user, password))
else:
self._driver = driver
self._graph_label = graph_label
self._typing_label = typing_label
self._relation_label = relation_label
self._graph_edge_label = graph_edge_label
self._graph_typing_label = graph_typing_label
self._graph_relation_label = graph_relation_label
try:
query = "CREATE " + constraint_query(
'n', self._graph_label, 'id')
self.execute(query)
except:
Dictionary contaning typing of data nodes by schema nodes.
By default is empty.
"""
if data_graph is None:
data_graph = {"nodes": [], "edges": []}
if schema_graph is None:
schema_graph = {"nodes": [], "edges": []}
if typing is None:
typing = dict()
if clear is True:
self._clear()
self._driver = GraphDatabase.driver(
uri, auth=(user, password))
self._graph_label = "graph"
self._typing_label = "homomorphism"
self._relation_label = None
self._graph_edge_label = "edge"
self._graph_typing_label = "typing"
self._graph_relation_label = "relation"
self._schema_node_label = "type"
self._data_node_label = "node"