Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import unittest
from neomodel import config, StructuredNode, StringProperty, install_all_labels
from neomodel.core import db
config.AUTO_INSTALL_LABELS = False
class NoConstraintsSetup(StructuredNode):
name = StringProperty(unique_index=True)
config.AUTO_INSTALL_LABELS = True
def test_labels_were_not_installed():
bob = NoConstraintsSetup(name='bob').save()
bob2 = NoConstraintsSetup(name='bob').save()
assert bob.id != bob2.id
for n in NoConstraintsSetup.nodes.all():
n.delete()
@unittest.skip('disabled, broken in travis')
def test_install_all():
# run install all labels
install_all_labels()
assert True
from __future__ import print_function
import warnings
import os
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
from __future__ import print_function
import warnings
import os
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
def read_settings(self):
config.DATABASE_URL = getattr(settings, 'NEOMODEL_NEO4J_BOLT_URL', config.DATABASE_URL)
config.FORCE_TIMEZONE = getattr(settings, 'NEOMODEL_FORCE_TIMEZONE', False)
config.ENCRYPTED_CONNECTION = getattr(settings, 'NEOMODEL_ENCRYPTED_CONNECTION', True)
config.MAX_POOL_SIZE = getattr(settings, 'NEOMODEL_MAX_POOL_SIZE', config.MAX_POOL_SIZE)
def create_app():
app = Flask(__name__)
@app.route('/')
def index():
return "Hello World"
neomodel.config.DATABASE_URL = global_config.DB_URL
neomodel.config.AUTO_INSTALL_LABELS = True
neomodel.config.FORCE_TIMEZONE = True # default False
if (global_config.LOG_ENABLED):
logging.basicConfig(filename=os.path.abspath(os.path.join(
global_config.LOG_LOCATION, global_config.LOG_FILENAME)), format=global_config.LOG_FORMAT)
app.ext_logger = logging.getLogger()
app.ext_logger.setLevel(global_config.LOG_LEVEL)
handler = logging.handlers.RotatingFileHandler(
os.path.abspath(os.path.join(
global_config.LOG_LOCATION, global_config.LOG_FILENAME)),
maxBytes=global_config.LOG_MAX_BYTES,
backupCount=global_config.LOG_BACKUP_COUNT)
app.ext_logger.addHandler(handler)
else:
app.ext_logger = app.logger
def read_settings(self):
config.DATABASE_URL = getattr(settings, 'NEOMODEL_NEO4J_BOLT_URL', config.DATABASE_URL)
config.FORCE_TIMEZONE = getattr(settings, 'NEOMODEL_FORCE_TIMEZONE', False)
config.ENCRYPTED_CONNECTION = getattr(settings, 'NEOMODEL_ENCRYPTED_CONNECTION', True)
config.MAX_POOL_SIZE = getattr(settings, 'NEOMODEL_MAX_POOL_SIZE', config.MAX_POOL_SIZE)
# from bulbs.utils import current_datetime
#
# class System(Node):
# name = String(nullable=False);
# # age = models.IntegerProperty()
#
# # neighbour = models.Relationship('self',rel_type='is_connected')
#
# class isConnected(Relationship):
# neighbour = DateTime(default=current_datetime, nullable=False)
from neomodel import (config, StructuredNode, StringProperty, IntegerProperty,
UniqueIdProperty, RelationshipTo, RelationshipFrom, Relationship)
config.DATABASE_URL = 'bolt://neo4j:Neo4j@localhost:7687'
class Machine(StructuredNode):
uid = UniqueIdProperty()
ip = StringProperty(unique_index=True)
subnet = StringProperty(unique_index=True)
hostname = StringProperty(unique_index=True)
tag = StringProperty(unique_index=True)
distance = IntegerProperty()
queue = IntegerProperty()
connected = Relationship('Machine','IS_CONNECTED')
# objects = StructuredNode.Manager()
# age = IntegerProperty(index=True, default=0)
def __str__(self):
return self.hostname
from django.apps import AppConfig
from django.conf import settings
from neomodel import config
config.AUTO_INSTALL_LABELS = False
class NeomodelConfig(AppConfig):
name = 'django_neomodel'
verbose_name = 'Django neomodel'
def read_settings(self):
config.DATABASE_URL = getattr(settings, 'NEOMODEL_NEO4J_BOLT_URL', config.DATABASE_URL)
config.FORCE_TIMEZONE = getattr(settings, 'NEOMODEL_FORCE_TIMEZONE', False)
config.ENCRYPTED_CONNECTION = getattr(settings, 'NEOMODEL_ENCRYPTED_CONNECTION', True)
config.MAX_POOL_SIZE = getattr(settings, 'NEOMODEL_MAX_POOL_SIZE', config.MAX_POOL_SIZE)
def ready(self):
self.read_settings()
def create_app():
app = Flask(__name__)
@app.route('/')
def index():
return "Hello World"
neomodel.config.DATABASE_URL = global_config.DB_URL
neomodel.config.AUTO_INSTALL_LABELS = True
neomodel.config.FORCE_TIMEZONE = True # default False
if global_config.LOG_ENABLED:
logging.basicConfig(filename=os.path.abspath(os.path.join(
global_config.LOG_LOCATION, global_config.LOG_FILENAME)), format=global_config.LOG_FORMAT)
app.ext_logger = logging.getLogger()
app.ext_logger.setLevel(global_config.LOG_LEVEL)
handler = logging.handlers.RotatingFileHandler(
os.path.abspath(os.path.join(
global_config.LOG_LOCATION, global_config.LOG_FILENAME)),
maxBytes=global_config.LOG_MAX_BYTES,
backupCount=global_config.LOG_BACKUP_COUNT)
app.ext_logger.addHandler(handler)
else:
app.ext_logger = app.logger
PersonsView.register(app, route_base="/persons", trailing_slash=False)
def create_app():
app = Flask(__name__)
@app.route('/')
def index():
return "Hello World"
neomodel.config.DATABASE_URL = global_config.DB_URL
neomodel.config.AUTO_INSTALL_LABELS = True
neomodel.config.FORCE_TIMEZONE = True # default False
if global_config.LOG_ENABLED:
logging.basicConfig(filename=os.path.abspath(os.path.join(
global_config.LOG_LOCATION, global_config.LOG_FILENAME)), format=global_config.LOG_FORMAT)
app.ext_logger = logging.getLogger()
app.ext_logger.setLevel(global_config.LOG_LEVEL)
handler = logging.handlers.RotatingFileHandler(
os.path.abspath(os.path.join(
global_config.LOG_LOCATION, global_config.LOG_FILENAME)),
maxBytes=global_config.LOG_MAX_BYTES,
backupCount=global_config.LOG_BACKUP_COUNT)
app.ext_logger.addHandler(handler)
else:
app.ext_logger = app.logger