Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from neomodel import StructuredNode, StringProperty
from neomodel.contrib import Multilingual, Language
class Student(Multilingual, StructuredNode):
name = StringProperty(unique_index=True)
def setup():
for l in ['fr', 'ar', 'pl', 'es']:
Language(code=l).save()
def test_multilingual():
bob = Student(name="Bob", age=77).save()
bob.attach_language(Language.get("fr"))
bob.attach_language("ar")
bob.attach_language(Language.get("ar"))
bob.attach_language(Language.get("pl"))
assert bob.has_language("fr")
assert not bob.has_language("es")
def test_default_value_callable():
def uid_generator():
return 'xx'
class DefaultTestValueTwo(StructuredNode):
uid = StringProperty(default=uid_generator, index=True)
a = DefaultTestValueTwo().save()
assert a.uid == 'xx'
users = Customer.create(
{'email': 'jim6@aol.com', 'age': 3},
)
assert users
with raises(UniqueProperty):
Customer.create(
{'email': 'jim6@aol.com', 'age': 3},
{'email': 'jim7@aol.com', 'age': 5},
)
# not found
assert not Customer.nodes.filter(email='jim7@aol.com')
class Dog(StructuredNode):
name = StringProperty(required=True)
owner = RelationshipTo('Person', 'owner')
class Person(StructuredNode):
name = StringProperty(unique_index=True)
pets = RelationshipFrom('Dog', 'owner')
def test_get_or_create_with_rel():
bob = Person.get_or_create({"name": "Bob"})[0]
bobs_gizmo = Dog.get_or_create({"name": "Gizmo"}, relationship=bob.pets)
tim = Person.get_or_create({"name": "Tim"})[0]
tims_gizmo = Dog.get_or_create({"name": "Gizmo"}, relationship=tim.pets)
from pytest import raises
from neomodel import (StructuredNode, StringProperty, IntegerProperty, OneOrMore, ZeroOrMore,
RelationshipTo, AttemptedCardinalityViolation, CardinalityViolation,
ZeroOrOne, One)
class HairDryer(StructuredNode):
version = IntegerProperty()
class ScrewDriver(StructuredNode):
version = IntegerProperty()
class Car(StructuredNode):
version = IntegerProperty()
class Monkey(StructuredNode):
name = StringProperty()
dryers = RelationshipTo('HairDryer', 'OWNS_DRYER', cardinality=ZeroOrMore)
driver = RelationshipTo('ScrewDriver', 'HAS_SCREWDRIVER', cardinality=ZeroOrOne)
car = RelationshipTo('Car', 'HAS_CAR', cardinality=OneOrMore)
toothbrush = RelationshipTo('ToothBrush', 'HAS_TOOTHBRUSH', cardinality=One)
class ToothBrush(StructuredNode):
name = StringProperty()
def test_cardinality_zero_or_more():
version = IntegerProperty()
class Car(StructuredNode):
version = IntegerProperty()
class Monkey(StructuredNode):
name = StringProperty()
dryers = RelationshipTo('HairDryer', 'OWNS_DRYER', cardinality=ZeroOrMore)
driver = RelationshipTo('ScrewDriver', 'HAS_SCREWDRIVER', cardinality=ZeroOrOne)
car = RelationshipTo('Car', 'HAS_CAR', cardinality=OneOrMore)
toothbrush = RelationshipTo('ToothBrush', 'HAS_TOOTHBRUSH', cardinality=One)
class ToothBrush(StructuredNode):
name = StringProperty()
def test_cardinality_zero_or_more():
m = Monkey(name='tim').save()
assert m.dryers.all() == []
assert m.dryers.single() is None
h = HairDryer(version=1).save()
m.dryers.connect(h)
assert len(m.dryers.all()) == 1
assert m.dryers.single().version == 1
m.dryers.disconnect(h)
assert m.dryers.all() == []
assert m.dryers.single() is None
def test_save_through_magic_property():
user = User(email_alias='blah@test.com', age=8).save()
assert user.email_alias == 'blah@test.com'
user = User.nodes.get(email='blah@test.com')
assert user.email == 'blah@test.com'
assert user.email_alias == 'blah@test.com'
user1 = User(email='blah1@test.com', age=8).save()
assert user1.email_alias == 'blah1@test.com'
user1.email_alias = 'blah2@test.com'
assert user1.save()
user2 = User.nodes.get(email='blah2@test.com')
assert user2
class Customer2(StructuredNode):
__label__ = 'customers'
email = StringProperty(unique_index=True, required=True)
age = IntegerProperty(index=True)
def test_not_updated_on_unique_error():
Customer2(email='jim@bob.com', age=7).save()
test = Customer2(email='jim1@bob.com', age=2).save()
test.email = 'jim@bob.com'
with raises(UniqueProperty):
test.save()
customers = Customer2.nodes.all()
assert customers[0].email != customers[1].email
assert Customer2.nodes.get(email='jim@bob.com').age == 7
assert Customer2.nodes.get(email='jim1@bob.com').age == 2
def test_independent_property_name_get_or_create():
class TestNode(StructuredNode):
uid = UniqueIdProperty()
name_ = StringProperty(db_property="name", required=True)
# create the node
TestNode.get_or_create({'uid': 123, 'name_': 'jim'})
# test that the node is retrieved correctly
x = TestNode.get_or_create({'uid': 123, 'name_': 'jim'})[0]
# check database property name on low level
results, meta = db.cypher_query("MATCH (n:TestNode) RETURN n")
node_properties = _get_node_properties(results[0][0])
assert node_properties['name'] == "jim"
assert 'name_' not in node_properties
# delete node afterwards
x.delete()
def add_locale(self, lang):
if not isinstance(lang, StructuredNode):
lang = Locale.get(lang)
self.locales.connect(lang)
from neomodel import (
StringProperty,
StructuredNode,
RelationshipTo,
Relationship
)
from .nodeutils import NodeUtils
class Intermediary(StructuredNode, NodeUtils):
sourceID = StringProperty()
valid_until = StringProperty()
name = StringProperty()
country_codes = StringProperty()
countries = StringProperty()
node_id = StringProperty(index = True)
status = StringProperty()
entities = RelationshipTo('.entity.Entity', 'INTERMEDIARY_OF')
addresses = RelationshipTo('.address.Address', 'REGISTERED_ADDRESS')
officers = Relationship('.officer.Officer', None)
@property
def serialize(self):
return {
'node_properties': {
cursor = connection.cursor()
cursor.execute(
'REFRESH MATERIALIZED VIEW CONCURRENTLY grafit_search_word;')
logger.info("finished updating search word")
class ArticleRel(StructuredRel):
created_at = DateTimeProperty(
default=lambda: datetime.now()
)
tf_idf = FloatProperty()
hidden = BooleanProperty(default=False)
label = StringProperty()
class GraphArticle(StructuredNode):
uid = UniqueIdProperty()
name = StringProperty()
related = Relationship('GraphArticle', 'RELATED', model=ArticleRel)
class SearchResult(models.Model):
id = models.BigIntegerField(primary_key=True)
title = models.TextField()
headline = models.TextField()
rank = models.DecimalField(max_digits=19, decimal_places=2)
class Meta:
managed = False
db_table = 'search_index'