Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
unconstrained_property = StringProperty()
# Create a node with a missing required property
with raises(RequiredProperty):
x = ConstrainedTestNode(required_property="required", unique_property="unique")
x.save()
# Create a node with a missing unique (but not required) property.
x = ConstrainedTestNode()
x.required_property = "required"
x.unique_required_property = "unique and required"
x.unconstrained_property = "no contraints"
x.save()
# check database property name on low level
results, meta = db.cypher_query("MATCH (n:ConstrainedTestNode) RETURN n")
node_properties = _get_node_properties(results[0][0])
assert node_properties["unique_required_property"] == "unique and required"
# delete node afterwards
x.delete()
def test_independent_property_name():
class TestDBNamePropertyNode(StructuredNode):
name_ = StringProperty(db_property="name")
x = TestDBNamePropertyNode()
x.name_ = "jim"
x.save()
# check database property name on low level
results, meta = db.cypher_query("MATCH (n:TestDBNamePropertyNode) RETURN n")
node_properties = _get_node_properties(results[0][0])
assert node_properties['name'] == "jim"
node_properties = _get_node_properties(results[0][0])
assert not 'name_' in node_properties
assert not hasattr(x, 'name')
assert hasattr(x, 'name_')
assert TestDBNamePropertyNode.nodes.filter(name_="jim").all()[0].name_ == x.name_
assert TestDBNamePropertyNode.nodes.get(name_="jim").name_ == x.name_
x.delete()
# Nameless
x = UniqueNullableNameNode()
x.save()
y = UniqueNullableNameNode()
y.save()
# Named
z = UniqueNullableNameNode(name="named")
z.save()
with raises(UniqueProperty):
a = UniqueNullableNameNode(name="named")
a.save()
# Check nodes are in database
results, meta = db.cypher_query("MATCH (n:UniqueNullableNameNode) RETURN n")
assert len(results) == 3
# Delete nodes afterwards
x.delete()
y.delete()
z.delete()
def test_recursive_automatic_object_resolution():
"""
Node objects are instantiated to native Python objects, both at the top
level of returned results and in the case where they are returned within
lists.
"""
# Create a few entities
A = TechnicalPerson.get_or_create({"name":"Grumpier", "expertise":"Grumpiness"})[0]
B = TechnicalPerson.get_or_create({"name":"Happier", "expertise":"Grumpiness"})[0]
C = TechnicalPerson.get_or_create({"name":"Sleepier", "expertise":"Pillows"})[0]
D = TechnicalPerson.get_or_create({"name":"Sneezier", "expertise":"Pillows"})[0]
# Retrieve mixed results, both at the top level and nested
L, _ = neomodel.db.cypher_query("MATCH (a:TechnicalPerson) "
"WHERE a.expertise='Grumpiness' "
"WITH collect(a) as Alpha "
"MATCH (b:TechnicalPerson) "
"WHERE b.expertise='Pillows' "
"WITH Alpha, collect(b) as Beta "
"RETURN [Alpha, [Beta, [Beta, ['Banana', "
"Alpha]]]]", resolve_objects = True)
# Assert that a Node returned deep in a nested list structure is of the
# correct type
assert type(L[0][0][0][1][0][0][0][0]) is TechnicalPerson
# Assert that primitive data types remain primitive data types
assert issubclass(type(L[0][0][0][1][0][1][0][1][0][0]), basestring)
A.delete()
B.delete()
def delete_all(self, request):
try:
# patch __log
self.__log = self._GRest__log
(primary, _) = validate_models(self)
if all([ENABLE_DELETE_ALL == "True", primary.model]):
# user wants to delete all items (including relations)
results = db.cypher_query("MATCH (n:{0}) DETACH DELETE n".format(
primary.model.__name__))
if results[0] == []:
return serialize(dict(result="OK"))
else:
raise HTTPException(msg.DELETE_FAILED, 500)
else:
raise HTTPException(msg.FEATURE_IS_DISABLED, 403)
except (DoesNotExist, AttributeError) as e:
self.__log.exception(e)
raise HTTPException(msg.ITEM_DOES_NOT_EXIST, 404)
except UniqueProperty as e:
self.__log.exception(e)
raise HTTPException(msg.NON_UNIQUE_PROPERTY, 409)
except RequiredProperty as e:
self.__log.exception(e)
raise HTTPException(msg.REQUIRE_PROPERTY_MISSING, 500)
from neomodel import db
countries = db.cypher_query(
'''
MATCH (n)
WHERE NOT n.countries CONTAINS ';'
RETURN DISTINCT n.countries AS countries
'''
)[0]
jurisdictions = db.cypher_query(
'''
MATCH (n)
RETURN DISTINCT n.jurisdiction AS jurisdiction
'''
)[0]
data_sources = db.cypher_query(
'''
def get_all_countries():
query = "MATCH (n) WHERE NOT n.countries CONTAINS ';' RETURN DISTINCT 'node' as entity, n.countries AS countries UNION ALL MATCH ()-[r]-() WHERE EXISTS(r.countries) RETURN DISTINCT 'relationship' AS entity, r.countries AS countries"
results = db.cypher_query(query)
return results
from neomodel import db
countries = db.cypher_query(
'''
MATCH (n)
WHERE NOT n.countries CONTAINS ';'
RETURN DISTINCT n.countries AS countries
'''
)[0]
jurisdictions = db.cypher_query(
'''
MATCH (n)
RETURN DISTINCT n.jurisdiction AS jurisdiction
'''
)[0]
data_sources = db.cypher_query(
'''
MATCH (n)
RETURN DISTINCT n.sourceID AS dataSource
'''
)[0]
COUNTRIES = sorted([country[0] for country in countries])
JURISDICTIONS = sorted([jurisdiction[0] for jurisdiction in jurisdictions if isinstance(jurisdiction[0], str)])
DATASOURCE = sorted([data_source[0] for data_source in data_sources if isinstance(data_source[0], str)])