Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_rollback_and_commit_transaction():
for p in APerson.nodes:
p.delete()
APerson(name='Roger').save()
db.begin()
APerson(name='Terry S').save()
db.rollback()
assert len(APerson.nodes) == 1
db.begin()
APerson(name='Terry S').save()
db.commit()
assert len(APerson.nodes) == 2
@db.transaction
def in_a_tx(*names):
for n in names:
APerson(name=n).save()
class SuperTechnicalPerson(TechnicalPerson):
superness = neomodel.FloatProperty(default=1.0)
class UltraTechnicalPerson(SuperTechnicalPerson):
ultraness = neomodel.FloatProperty(default=3.1415928)
# Create a TechnicalPerson...
A = TechnicalPerson.get_or_create({"name":"Grumpy", "expertise":"Grumpiness"})[0]
# ...that is connected to an UltraTechnicalPerson
F = UltraTechnicalPerson(name = "Chewbaka", expertise="Aarrr wgh ggwaaah").save()
A.friends_with.connect(F)
# Forget about the UltraTechnicalPerson
del neomodel.db._NODE_CLASS_REGISTRY[frozenset(["UltraTechnicalPerson",
"SuperTechnicalPerson",
"TechnicalPerson",
"BasePerson"])]
# Recall a TechnicalPerson and enumerate its friends.
# One of them is UltraTechnicalPerson which would be returned as a valid
# node to a friends_with query but is currently unknown to the node class registry.
A = TechnicalPerson.get_or_create({"name":"Grumpy", "expertise":"Grumpiness"})[0]
with pytest.raises(neomodel.exceptions.ModelDefinitionMismatch):
for some_friend in A.friends_with:
print(some_friend.name)
def test_transaction_as_a_context():
with db.transaction:
APerson(name='Tim').save()
assert APerson.nodes.filter(name='Tim')
with raises(UniqueProperty):
with db.transaction:
APerson(name='Tim').save()
from neomodel import config, db, clear_neo4j_database, change_neo4j_password
from neo4j.v1 import CypherError
warnings.simplefilter('default')
config.DATABASE_URL = os.environ.get('NEO4J_BOLT_URL', 'bolt://neo4j:neo4j@localhost:7687')
config.AUTO_INSTALL_LABELS = True
try:
clear_neo4j_database(db)
except CypherError as ce:
# handle instance without password being changed
if 'The credentials you provided were valid, but must be changed before you can use this instance' in str(ce):
change_neo4j_password(db, 'test')
db.set_connection('bolt://neo4j:test@localhost:7687')
print("New database with no password set, setting password to 'test'")
print("Please 'export NEO4J_BOLT_URL=bolt://neo4j:test@localhost:7687' for subsequent test runs")
else:
raise ce
raise HTTPException(msg.RELATION_EXISTS, 409)
else:
# Get relation between primary and secondary objects
relation = getattr(
primary_selected_item,
secondary.model_name)
# If there is a relation model between the two,
# validate requests based on that
relation_model = relation.definition["model"]
json_data = {}
if relation_model is not None:
# TODO: find a way to validate relationships
json_data = request.get_json(silent=True)
with db.transaction:
if not json_data:
related_item = relation.connect(
secondary_selected_item)
else:
related_item = relation.connect(
secondary_selected_item, json_data)
if related_item:
return serialize(dict(result="OK"))
else:
raise HTTPException(msg.RELATION_DOES_NOT_EXIST,
404)
elif all([primary_selected_item is None,
secondary.model is None,
secondary.id is None]):
new_item = primary.model()
:return: An HTTP response object in accordance with the specification at \
http://jsonapi.org/format/#fetching-resources
"""
try:
if request_args.get('include'):
raise ParameterNotSupported
offset = request_args.get('page[offset]', 0)
limit = request_args.get('page[limit]', 20)
query = "MATCH (n) WHERE n:{label} AND n.active RETURN n ORDER BY n.id SKIP {offset} LIMIT {limit}".format(
label=cls.__name__,
offset=offset,
limit=limit)
results, meta = db.cypher_query(query)
data = dict()
data['data'] = list()
data['links'] = dict()
data['links']['self'] = "{class_link}?page[offset]={offset}&page[limit]={limit}".format(
class_link=cls.get_class_link(),
offset=offset,
limit=limit
)
data['links']['first'] = "{class_link}?page[offset]={offset}&page[limit]={limit}".format(
class_link=cls.get_class_link(),
offset=0,
limit=limit
)