Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from pytest import raises
from neomodel import (StructuredNode, StringProperty, IntegerProperty, UniqueIdProperty,
RelationshipTo, RelationshipFrom)
from neomodel.exceptions import UniqueProperty, DeflateError
class UniqueUser(StructuredNode):
uid = UniqueIdProperty()
name = StringProperty()
age = IntegerProperty()
def test_unique_id_property_batch():
users = UniqueUser.create(
{'name': 'bob', 'age': 2},
{'name': 'ben', 'age': 3}
)
assert users[0].uid != users[1].uid
users = UniqueUser.get_or_create(
{'uid': users[0].uid},
{'name': 'bill', 'age': 4}
)
from datetime import datetime, timedelta
twelve_days = timedelta(days=12)
eleven_days = timedelta(days=11)
ten_days = timedelta(days=10)
nine_days = timedelta(days=9)
now = datetime.now()
class FriendRelationship(StructuredRel):
since = DateTimeProperty(default=datetime.now)
class Person(StructuredNode):
name = StringProperty()
age = IntegerProperty()
friends = Relationship('Person', 'friend_of', model=FriendRelationship)
def setup_friends(person0, person1, since=None):
rel = person0.friends.connect(person1)
if (since):
rel.since = since
rel.save()
return rel.since
def test_traversal_single_param():
connection().clear()
jean = Person(name="Jean", age=25).save()
johan = Person(name="Johan", age=19).save()
from pytest import raises
from neomodel import StructuredNode, StringProperty, IntegerProperty, UniqueProperty
class Human(StructuredNode):
name = StringProperty(unique_index=True)
age = IntegerProperty(index=True)
def test_unique_error():
Human(name="j1m", age=13).save()
try:
Human(name="j1m", age=14).save()
except UniqueProperty as e:
assert True
assert str(e).find('j1m')
assert str(e).find('name')
assert str(e).find('FooBarr')
else:
assert False, "UniqueProperty not raised."
def pre_save(self):
HOOKS_CALLED['pre_save'] += 1
def post_save(self):
HOOKS_CALLED['post_save'] += 1
class Badger(StructuredNode):
name = StringProperty(unique_index=True)
friend = Relationship('Badger', 'FRIEND', model=FriendRel)
hates = RelationshipTo('Stoat', 'HATES', model=HatesRel)
class Stoat(StructuredNode):
name = StringProperty(unique_index=True)
hates = RelationshipTo('Badger', 'HATES', model=HatesRel)
def test_either_connect_with_rel_model():
paul = Badger(name="Paul").save()
tom = Badger(name="Tom").save()
# creating rels
new_rel = tom.friend.disconnect(paul)
new_rel = tom.friend.connect(paul)
assert isinstance(new_rel, FriendRel)
assert isinstance(new_rel.since, datetime)
# updating properties
new_rel.since = datetime.now(pytz.utc)
assert isinstance(new_rel.save(), FriendRel)
StringProperty,
StructuredNode,
RelationshipTo,
RelationshipFrom,
Relationship
)
from .nodeutils import NodeUtils
class Entity(StructuredNode, NodeUtils):
sourceID = StringProperty()
address = StringProperty()
jurisdiction = StringProperty()
service_provider = StringProperty()
countries = StringProperty()
jurisdiction_description = StringProperty()
valid_until = StringProperty()
ibcRUC = StringProperty()
name = StringProperty()
country_codes = StringProperty()
incorporation_date = StringProperty()
node_id = StringProperty(index = True)
status = StringProperty()
officers = RelationshipFrom('.officer.Officer', 'OFFICER_OF')
intermediaries = RelationshipFrom('.intermediary.Intermediary', 'INTERMEDIARY_OF')
addresses = RelationshipTo('.address.Address', 'REGISTERED_ADDRESS')
others = RelationshipFrom('.other.Other', 'CONNECTED_TO')
entities = Relationship('.entity.Entity', None)
@property
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': {
'sourceID': self.sourceID,
'valid_until': self.valid_until,
quotes = neomodel.Relationship('Tweet', 'QUOTES')
def save(self):
self.modified = datetime.datetime.now()
super(Tweet, self).save()
class User(neomodel.StructuredNode):
id_str = neomodel.StringProperty(unique_index=True, required=True)
name = neomodel.StringProperty(required=False)
screen_name = neomodel.StringProperty(required=False)
followers_count = neomodel.IntegerProperty(required=False)
friends_count = neomodel.IntegerProperty(required=False)
modified = neomodel.DateTimeProperty(required=False)
created_at = neomodel.DateTimeProperty(required=False)
description = neomodel.StringProperty(required=False)
location = neomodel.StringProperty(required=False)
coordinates = neomodel.ArrayProperty(required=False, default=[])
time_zone = neomodel.StringProperty(required=False)
url = neomodel.StringProperty(required=False)
lang = neomodel.StringProperty(required=False)
follows = neomodel.RelationshipTo('User', 'FOLLOWS')
posts = neomodel.RelationshipTo('Tweet', 'POSTS')
def save(self):
self.modified = datetime.datetime.now()
super(User, self).save()
class Hashtag(neomodel.StructuredNode):
text = neomodel.StringProperty(unique_index=True, required=True)
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': {
'sourceID': self.sourceID,
'valid_until': self.valid_until,
'name': self.name,
'country_codes': self.country_codes,
'countries': self.countries,
'node_id': self.node_id,
'status': self.status,
return None
class EnumeratedTypeError(Exception):
pass
class SerializableStructuredRel(StructuredRel):
r"""
The Base Relationship that all Structured Relationships must inherit from. All relationships should be structured \
starting version 1.1.0 -- okay to use model=SerializableStructuredRel
"""
secret = []
updated = DateTimeProperty(default=datetime.now())
created = DateTimeProperty(default=datetime.now())
type = StringProperty(default="serializable_structured_rel")
def get_resource_identifier_object(self, end_node):
try:
response = dict()
response['id'] = end_node.id
response['type'] = end_node.type
response['meta'] = dict()
props = self.defined_properties()
print self.__class__
for attr_name in props.keys():
print attr_name
if attr_name not in self.secret:
response['meta'][attr_name] = getattr(self, attr_name)
return response
def create_resources(resource_types):
for resource_name, resource in resource_types.items():
fields = {
'uid': StringProperty(unique_index=True),
'name': StringProperty(required=True),
'kind': StringProperty(required=True),
'status': StringProperty(default='Unknown'),
'metadata': JSONProperty(),
}
for field_name, field in resource.get('model', {}).items():
cls_name = field.pop("type")
target_cls = field.pop('target')
model_name = field.pop('model')
field['model'] = registry.get_type(model_name)
fields[field_name] = globals().get(utils.to_camel_case(cls_name))(target_cls, model_name, **field)
registry.add(type(resource_name,
(StructuredNode,), fields))