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_unique_path_not_unique_raises_cypher_transaction_error_in_transaction(self):
tx = self.graph.begin()
cursor = tx.run("CREATE (a), (b) RETURN a, b")
tx.process()
record = next(cursor)
parameters = {"A": record["a"].identity, "B": record["b"].identity}
statement = ("MATCH (a) WHERE id(a)={A} MATCH (b) WHERE id(b)={B}" +
"CREATE (a)-[:KNOWS]->(b)")
tx.run(statement, parameters)
tx.run(statement, parameters)
statement = ("MATCH (a) WHERE id(a)={A} MATCH (b) WHERE id(b)={B}" +
"CREATE UNIQUE (a)-[:KNOWS]->(b)")
with self.assertRaises(ConstraintError):
tx.run(statement, parameters)
tx.commit()
def test_unique_path_not_unique_raises_cypher_transaction_error_in_transaction(self):
tx = self.graph.begin()
cursor = tx.run("CREATE (a), (b) RETURN a, b")
tx.process()
record = cursor.next()
parameters = {"A": record["a"].identity, "B": record["b"].identity}
statement = ("MATCH (a) WHERE id(a)={A} MATCH (b) WHERE id(b)={B}" +
"CREATE (a)-[:KNOWS]->(b)")
tx.run(statement, parameters)
tx.run(statement, parameters)
statement = ("MATCH (a) WHERE id(a)={A} MATCH (b) WHERE id(b)={B}" +
"CREATE UNIQUE (a)-[:KNOWS]->(b)")
with self.assertRaises(ConstraintError):
tx.run(statement, parameters)
tx.commit()
def test_unique_constraint(self):
label_1 = next(self.unique_string)
borough = Node(label_1, name="Taufkirchen")
self.graph.create(borough)
self.schema.create_uniqueness_constraint(label_1, "name")
constraints = self.schema.get_uniqueness_constraints(label_1)
assert (u"name",) in constraints
with self.assertRaises(ConstraintError):
self.graph.create(Node(label_1, name="Taufkirchen"))
self.graph.delete(borough)
def test_unique_path_not_unique_raises_cypher_error(self):
graph = self.graph
record = graph.run("CREATE (a), (b) RETURN a, b").next()
parameters = {"A": record["a"].identity, "B": record["b"].identity}
statement = ("MATCH (a) WHERE id(a)={A} MATCH (b) WHERE id(b)={B}"
"CREATE (a)-[:KNOWS]->(b)")
graph.run(statement, parameters)
graph.run(statement, parameters)
statement = ("MATCH (a) WHERE id(a)={A} MATCH (b) WHERE id(b)={B}"
"CREATE UNIQUE (a)-[:KNOWS]->(b)")
with self.assertRaises(ConstraintError):
graph.run(statement, parameters)
if holistic:
with self._driver.session() as session:
tx = session.begin_transaction()
load_graph_from_json_apoc(
tx, json_data, graph_id, self._graph_edge_label)
tx.commit()
else:
g = self.get_graph(graph_id)
for n in json_data["nodes"]:
g.add_node(n["id"], attrs_from_json(n["attrs"]))
for e in json_data["edges"]:
g.add_edge(
e["from"], e["to"],
attrs_from_json(e["attrs"]))
except(ConstraintError):
raise HierarchyError(
"The graph '{}' is already in the database.".format(graph_id))
class DatabaseError(GraphError):
""" The database failed to service the request.
"""
class TransientError(GraphError):
""" The database cannot service the request right now, retrying later might yield a successful outcome.
"""
client_errors = {
# ConstraintError
"Neo.ClientError.Schema.ConstraintValidationFailed": ConstraintError,
"Neo.ClientError.Schema.ConstraintViolation": ConstraintError,
"Neo.ClientError.Statement.ConstraintVerificationFailed": ConstraintError,
"Neo.ClientError.Statement.ConstraintViolation": ConstraintError,
# CypherSyntaxError
"Neo.ClientError.Statement.InvalidSyntax": CypherSyntaxError,
"Neo.ClientError.Statement.SyntaxError": CypherSyntaxError,
# CypherTypeError
"Neo.ClientError.Procedure.TypeError": CypherTypeError,
"Neo.ClientError.Statement.InvalidType": CypherTypeError,
"Neo.ClientError.Statement.TypeError": CypherTypeError,
# Forbidden
"Neo.ClientError.General.ForbiddenOnReadOnlyDatabase": Forbidden,
"Neo.ClientError.General.ReadOnly": Forbidden,
"Neo.ClientError.Schema.ForbiddenOnConstraintIndex": Forbidden,
If graph with provided id already exists in the hierarchy
"""
try:
# Create a node in the hierarchy
query = "CREATE ({}:{} {{ id : '{}' }}) \n".format(
'new_graph',
self._graph_label,
graph_id)
if attrs is not None:
normalize_attrs(attrs)
query += set_attributes(
var_name='new_graph',
attrs=attrs)
self.execute(query)
except(ConstraintError):
raise HierarchyError(
"The graph '{}' is already in the database.".format(graph_id))
g = graphs.Neo4jGraph(
driver=self._driver,
node_label=graph_id,
unique_node_ids=True)
if node_list is not None:
g.add_nodes_from(node_list)
if edge_list is not None:
g.add_edges_from(edge_list)
If graph with provided id already exists in the hierarchy
"""
try:
# Create a node in the hierarchy
query = "CREATE ({}:{} {{ id : '{}' }}) \n".format(
'new_graph',
self._graph_label,
graph_id)
if attrs is not None:
normalize_attrs(attrs)
query += set_attributes(
var_name='new_graph',
attrs=attrs)
self.execute(query)
except(ConstraintError):
raise HierarchyError(
"The graph '{}' is already in the database.".format(graph_id))
g = graphs.Neo4jGraph(
driver=self._driver,
node_label=graph_id,
unique_node_ids=True)
if node_list is not None:
g.add_nodes_from(node_list)
if edge_list is not None:
g.add_edges_from(edge_list)
graph_attrs : dict, optional
Dictionary containing attributes of the new node
"""
try:
# Create a node in the hierarchy
query = "CREATE ({}:{} {{ id : '{}' }}) \n".format(
'new_graph',
self._graph_label,
graph_id)
if attrs is not None:
normalize_attrs(attrs)
query += set_attributes(
var_name='new_graph',
attrs=attrs)
self.execute(query)
except(ConstraintError):
raise HierarchyError(
"The graph '{}' is already in the database.".format(graph_id))
g = Neo4jGraph(
driver=self._driver,
node_label=graph_id,
unique_node_ids=True)
if node_list is not None:
g.add_nodes_from(node_list)
if edge_list is not None:
g.add_edges_from(edge_list)