Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __properties__(self):
node_props = {}
for key, value in items(super(PropertyManager, self).__dict__):
if not (key.startswith('_') or value is None
or isinstance(value,
(types.MethodType, RelationshipManager, AliasProperty,))):
node_props[key] = value
return node_props
filter(lambda prop:
any([getattr(self, prop) is None,
isinstance(getattr(self, prop),
relationship_manager.RelationshipManager),
prop in blocked_properties]), properties))
from neomodel.exceptions import (
AttemptedCardinalityViolation, CardinalityViolation
)
from neomodel.relationship_manager import (
RelationshipManager, ZeroOrMore
) #noqa: F401
class ZeroOrOne(RelationshipManager):
""" A relationship to zero or one node. """
description = "zero or one relationship"
def single(self):
"""
Return the associated node.
:return: node
"""
nodes = super(ZeroOrOne, self).all()
if len(nodes) == 1:
return nodes[0]
if len(nodes) > 1:
raise CardinalityViolation(self, len(nodes))
def all(self):
def __init__(self, relation_type, cls_name, direction,
manager=RelationshipManager,
model=None):
self._raw_class = cls_name
self.manager = manager
self.definition = {}
self.definition['relation_type'] = relation_type
self.definition['direction'] = direction
self.definition['model'] = model
def __properties__(self):
from .relationship_manager import RelationshipManager
return dict((name, value) for name, value in vars(self).items()
if not name.startswith('_')
and not callable(value)
and not isinstance(value,
(RelationshipManager, AliasProperty,))
)
batch.add_indexed_node(cls.index.__index__, key, value, node)
return batch
class CategoryNode(CypherMixin):
def __init__(self, name):
self.name = name
def traverse(self, rel):
return TraversalSet(self).traverse(rel)
def _pre_action_check(self, action):
pass
class InstanceManager(RelationshipManager):
"""Manage 'instance' rel of category nodes"""
def connect(self, node):
raise Exception("connect not available from category node")
def disconnect(self, node):
raise Exception("disconnect not available from category node")
def category_factory(instance_cls):
""" Retrieve category node by name """
name = instance_cls.__name__
category_index = connection().get_or_create_index(neo4j.Node, 'Category')
category = CategoryNode(name)
category.__node__ = category_index.get_or_create('category', name, {'category': name})
rel_type = camel_to_upper(instance_cls.__name__)
category.instance = InstanceManager({
if nodes:
return nodes
raise CardinalityViolation(self, 'none')
def disconnect(self, node):
"""
Disconnect node
:param node:
:return:
"""
if super(OneOrMore, self).__len__() < 2:
raise AttemptedCardinalityViolation("One or more expected")
return super(OneOrMore, self).disconnect(node)
class One(RelationshipManager):
"""
A relationship to a single node
"""
description = "one relationship"
def single(self):
"""
Return the associated node.
:return: node
"""
nodes = super(One, self).all()
if nodes:
if len(nodes) == 1:
return nodes[0]
else:
self.definition['direction'] = direction
self.definition['model'] = model
def _lookup_node_class(self):
if not isinstance(self._raw_class, str):
self.definition['node_class'] = self._raw_class
else:
name = self._raw_class
self.definition['node_class'] = registry.get_type(name)
def build_manager(self, source, name):
self._lookup_node_class()
return self.manager(source, name, self.definition)
class ZeroOrMore(RelationshipManager):
description = "zero or more relationships"
def _relate(cls_name, direction, rel_type, cardinality=None, model=None):
if model and not issubclass(model, (StructuredRel,)):
raise ValueError('model must be a StructuredRel')
return RelationshipDefinition(rel_type,
cls_name,
direction,
cardinality,
model)
def RelationshipTo(cls_name, rel_type, cardinality=ZeroOrMore, model=None):
return _relate(cls_name, OUTGOING, rel_type, cardinality, model)
Connect to a node.
:param node:
:type: StructuredNode
:param properties: relationship properties
:type: dict
:return: True / rel instance
"""
if len(self):
raise AttemptedCardinalityViolation(
"Node already has {0} can't connect more".format(self))
else:
return super(ZeroOrOne, self).connect(node, properties)
class OneOrMore(RelationshipManager):
""" A relationship to zero or more nodes. """
description = "one or more relationships"
def single(self):
"""
Fetch one of the related nodes
:return: Node
"""
nodes = super(OneOrMore, self).all()
if nodes:
return nodes[0]
raise CardinalityViolation(self, 'none')
def all(self):
"""