Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_can_delete_property_on_node_in_same_batch():
graph_db = neo4j.GraphDatabaseService()
batch = neo4j.WriteBatch(graph_db)
alice = batch.create({"name": "Alice", "age": 34})
batch.delete_property(alice, "age")
results = batch.submit()
alice = results[batch.find(alice)]
assert alice["name"] == "Alice"
assert alice["age"] is None
def test_can_add_labels_to_node_in_same_batch():
graph_db = neo4j.GraphDatabaseService()
batch = neo4j.WriteBatch(graph_db)
alice = batch.create({"name": "Alice"})
batch.add_labels(alice, "human", "female")
batch.remove_label(alice, "female")
results = batch.submit()
alice = results[batch.find(alice)]
assert alice.get_labels() == {"human"}
def test_can_create_path_with_existing_nodes():
graph_db = neo4j.GraphDatabaseService()
alice, bob = graph_db.create({"name": "Alice"}, {"name": "Bob"})
batch = neo4j.WriteBatch(graph_db)
batch.create_path(alice, "KNOWS", bob)
results = batch.submit()
path = results[0]
assert len(path) == 1
assert path.nodes[0] == alice
assert path.relationships[0].type == "KNOWS"
assert path.nodes[1] == bob
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with the Assimilation Project software. If not, see http://www.gnu.org/licenses/
#
#
from __future__ import print_function
import re
# Import Neo4j modules
from py2neo import neo4j, cypher
# Attach to the graph db instance
graph_db = neo4j.GraphDatabaseService("http://localhost:7474/db/data/")
print("Version of Neo4J:", graph_db.neo4j_version)
# List of node types along with True if we create a corresponding index
nodetypes = {
"Ring": True,
"Drone": True,
"Switch": True,
"NIC": False,
"IPaddr": True,
"MACaddr": True,
}
indices = [key for key in nodetypes.keys() if nodetypes[key]]
def test_can_delete_properties_on_preexisting_node():
graph_db = neo4j.GraphDatabaseService()
alice, = graph_db.create({"name": "Alice", "age": 34})
batch = neo4j.WriteBatch(graph_db)
batch.delete_properties(alice)
batch.run()
props = alice.get_properties()
assert props == {}
def get_clean_database():
graph_db = neo4j.GraphDatabaseService()
if not graph_db.supports_schema_indexes:
return None
# Cleanup the database
# Constraints have to be removed before the indexed property keys can be removed.
graph_db.clear()
for label in graph_db.node_labels:
for key in graph_db.schema.get_unique_constraints(label):
graph_db.schema.remove_unique_constraint(label, key)
for key in graph_db.schema.get_indexed_property_keys(label):
graph_db.schema.drop_index(label, key)
return graph_db
#!/usr/bin/env python
import sys
from py2neo import neo4j
if len(sys.argv) < 2:
sys.exit("Usage: %s " % (sys.argv[0]))
# Hook into the database
gdb = neo4j.GraphDatabaseService("http://localhost:7474/db/data/")
root = gdb.get_subreference_node("NUMBERS")
# Set up pagination variables
page_number = int(sys.argv[1])
page_size = 10
# Build the cypher query
query = """\
start x=%s
match (x)-[:NUMBER]->(n)
return n
order by n.name
skip %d
limit %d
""" % (
str(root),
def __init__(self, conf_settings):
"""Constructor, setup neo4j graph database connection"""
self.conf_settings = conf_settings
self.db_conn = neo4j.GraphDatabaseService(str(self.conf_settings['neo4j_uri']))
self.indexes = list()
self.set_database_defaults()
#!/usr/bin/env python
"""
Example showing persistent objects
"""
from py2neo import neo4j
gdb = neo4j.GraphDatabaseService("http://localhost:7474/db/data")
# Define a few subclasses of PersistentObject
class Continent(neo4j.PersistentObject):
def __init__(self, node, name):
neo4j.PersistentObject.__init__(self, node)
self.name = name
class Country(neo4j.PersistentObject):
def __init__(self, node, name, population):
neo4j.PersistentObject.__init__(self, node)
self.name = name
self.population = population
self.currency = None
def find_shortest_path(node1, node2):
"""Connects to graph database, then creates and sends query to graph
database. Returns the shortest path between two nodes.
Format: (67149)-[:'LINKS_TO']->(421)"""
graph_db = neo4j.GraphDatabaseService()
t0 = time.time()
query = neo4j.CypherQuery(
graph_db,
"""MATCH (m:Page {node:{n1}}), (n:Page {node:{n2}}),
p = shortestPath((m)-[*..10]->(n)) RETURN p"""
)
try:
path = query.execute_one(n1=node1, n2=node2)
except:
path = None
t1 = time.time()
print "\nShortest Path:", path