Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from datetime import datetime
from pytest import raises
from neomodel import (
INCOMING, DateTimeProperty, IntegerProperty, RelationshipFrom, RelationshipTo,
StringProperty, StructuredNode, StructuredRel
)
from neomodel.match import NodeSet, QueryBuilder, Traversal
from neomodel.exceptions import MultipleNodesReturned
class SupplierRel(StructuredRel):
since = DateTimeProperty(default=datetime.now)
courier = StringProperty()
class Supplier(StructuredNode):
name = StringProperty()
delivery_cost = IntegerProperty()
coffees = RelationshipTo('Coffee', 'COFFEE SUPPLIERS') # Space to check for escaping
class Coffee(StructuredNode):
name = StringProperty(unique_index=True)
price = IntegerProperty()
suppliers = RelationshipFrom(Supplier, 'COFFEE SUPPLIERS', model=SupplierRel)
def test_filter_exclude_via_labels():
from datetime import datetime
from pytest import raises
import pytz
from neomodel import (StructuredNode, StructuredRel, Relationship, RelationshipTo,
StringProperty, DateTimeProperty, DeflateError)
HOOKS_CALLED = {
'pre_save': 0,
'post_save': 0
}
class FriendRel(StructuredRel):
since = DateTimeProperty(default=lambda: datetime.now(pytz.utc))
class HatesRel(FriendRel):
reason = StringProperty()
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)
from neomodel import (StructuredNode, StructuredRel, Relationship,
StringProperty, DateTimeProperty, IntegerProperty, connection)
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
class Meta:
app_label = 'someapp'
class Book(DjangoNode):
uid = UniqueIdProperty()
condition = StringProperty(default='new') # check fields can be omitted
format = StringProperty(required=True) # check required field can be ommitted on update
title = StringProperty(unique_index=True)
status = StringProperty(choices=(
('Available', 'A'),
('On loan', 'L'),
('Damaged', 'D'),
), default='Available')
created = DateTimeProperty(default=datetime.utcnow)
class Meta:
app_label = 'someapp'
tags = neomodel.RelationshipTo('Hashtag', 'TAGS')
contains = neomodel.RelationshipTo('Link', 'CONTAINS')
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()
import datetime
import neomodel
class Tweet(neomodel.StructuredNode):
id_str = neomodel.StringProperty(unique_index=True, required=True)
created_at = neomodel.DateTimeProperty(required=False)
modified = neomodel.DateTimeProperty(required=False)
retweeted = neomodel.BooleanProperty(required=False)
retweet_id_str = neomodel.StringProperty(required=False, default='')
reply_id_str = neomodel.StringProperty(required=False, default='')
quote_id_str = neomodel.StringProperty(required=False, default='')
mention_ids_str = neomodel.ArrayProperty(required=False, default=[])
text = neomodel.StringProperty(required=False)
coordinates = neomodel.ArrayProperty(required=False, default=[])
lang = neomodel.StringProperty(required=False)
features = neomodel.JSONProperty(required=False, default={})
retweets = neomodel.RelationshipTo('Tweet', 'RETWEETS')
mentions = neomodel.RelationshipTo('User', 'MENTIONS')
replies = neomodel.RelationshipTo('Tweet', 'REPLIES')
tags = neomodel.RelationshipTo('Hashtag', 'TAGS')
contains = neomodel.RelationshipTo('Link', 'CONTAINS')
return the_cls
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)
cursor.execute(
'REFRESH MATERIALIZED VIEW CONCURRENTLY grafit_search_index;')
logger.info("finished updating search index")
@receiver([signals.post_save, signals.post_delete], sender=Article, dispatch_uid="update_search_word")
def update_search_word(sender, instance, **kwargs):
logger.info("update search word")
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()
if hasattr(self, "__validation_rules__"):
if len(self.__validation_rules__) > 0:
# there is a set of user-defined validation rules
return self.__validation_rules__
model_types = [
StringProperty, DateTimeProperty, DateProperty,
EmailProperty, BooleanProperty, UniqueIdProperty,
ArrayProperty, IntegerProperty, JSONProperty
]
model_mapping = {
IntegerProperty: fields.Int,
StringProperty: fields.Str,
BooleanProperty: fields.Bool,
DateTimeProperty: fields.DateTime,
DateProperty: fields.Date,
EmailProperty: fields.Email,
ArrayProperty: fields.List,
JSONProperty: fields.Dict,
UniqueIdProperty: fields.UUID
}
name = 0
value = 1
for field in self.defined_properties().items():
if field[name] not in self.__validation_rules__:
if type(field[value]) in model_types:
if isinstance(field[value], ArrayProperty):
if field[value].unique_index:
# what it contains: Array of *String*