Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if arguments.password is not None and arguments.passwordFile is not None:
print("Not allowed to specify both password_file and password command-line options", file=sys.stderr)
sys.exit()
password = None
if password_file_name is not None:
with open(password_file_name[0], 'r') as password_file:
password = password_file.readline().rstrip("\n")
if password is None:
password = arguments.password
if password is not None:
password = password
if password is None:
password = getpass.getpass("Please enter the Neo4j database password: ")
bolt = 'bolt://127.0.0.1:7687'
# print("user: " + username + "; password: " + password)
driver = neo4j.GraphDatabase.driver(bolt, auth=(username, password))
node_label_list = node_labels() + ['Base']
print("NOTE: Please make sure that the Neo4j database is not set to read-only", file=sys.stderr)
# Create Indexes on Node Properties
create_index(node_label_list, "category")
create_index(node_label_list, "category_label")
create_index(node_label_list, "deprecated")
create_index(node_label_list, "full_name")
create_index(node_label_list, "name")
create_index(node_label_list, "provided_by")
create_index(node_label_list, "replaced_by")
create_index(node_label_list, "update_date")
constraint(node_label_list)
driver.close()
def _run_cypher_query(cypher_query: str, kg='KG2') -> List[Dict[str, any]]:
rtxc = RTXConfiguration()
if kg == 'KG2':
rtxc.live = "KG2"
try:
driver = GraphDatabase.driver(rtxc.neo4j_bolt, auth=(rtxc.neo4j_username, rtxc.neo4j_password))
with driver.session() as session:
query_results = session.run(cypher_query).data()
driver.close()
except Exception:
tb = traceback.format_exc()
error_type, error, _ = sys.exc_info()
print(f"Encountered an error interacting with {kg} neo4j. {tb}")
return []
else:
return query_results
def main():
# 'neo4j' and 'zooma-solr' apply only inside the docker network, but are
# portable across hosts, and the preferred option
stackhost = {'neo': 'neo4j',
'solr': 'solr',
'zooma_solr': 'zooma-solr'}
driver = GraphDatabase.driver("bolt://%s:7687" % stackhost['neo'])
cypher_query = """
MATCH (a)-[:HAS_PROVENANCE]->(s:Source)
WHERE s.name = 'pdx-finder'
WITH a
MATCH (be:BiologicalEntity)<-[:HAS_BIO_ENTITY]-(a:Annotation)-[:HAS_PROPERTY]->(p:Property)
OPTIONAL MATCH (a)-[:HAS_SEMANTIC_TAG]->(st:SemanticTag)
RETURN distinct be.bioEntity, p.propertyType, p.propertyValue, st.semanticTag
ORDER BY be.bioEntity
;"""
session = driver.session()
print('Running Cypher query ...')
results = session.run (cypher_query)
def __init__(
self,
uri: str = Config.get("neo4j", "host"),
username: str = Config.get("neo4j", "username"),
password: str = Config.get("neo4j", "password"),
clear_database: bool = False,
*args,
**kwargs,
):
logger.info(f"Connecting to neo4j server at {uri}")
self.neo4j = GraphDatabase.driver(uri, auth=(username, password))
super().__init__(*args, **kwargs)
logger.info("Initialized Neo4j Backend")
self.batch_size = int(Config.get("neo4j", "batch_size"))
self.uri = uri
if clear_database:
logger.info("Wiping database")
with self.neo4j.session() as session:
session.write_transaction(lambda tx: tx.run("MATCH (n) DETACH DELETE n"))
def cypher(plpy, query, params, url, dbname, login, password):
"""
Make cypher query and return JSON result
"""
driver = GraphDatabase.driver( url, auth=(login, password), encrypted=False)
# Execute & retrieve neo4j data
if driver.supports_multi_db():
session = driver.session(database=(dbname or 'neo4j'))
else:
session = driver.session()
plpy.debug("Cypher function with query " + query + " and params " + str(params))
# Execute & retrieve neo4j data
try:
for record in session.run(query, ast.literal_eval(params)):
jsonResult = "{"
for key in record.keys():
if len(jsonResult) > 1:
jsonResult += ","
def _run_cypher_query(cypher_query: str, kp: str, log: Response) -> List[Dict[str, any]]:
rtxc = RTXConfiguration()
if kp == "KG2": # Flip into KG2 mode if that's our KP (rtx config is set to KG1 info by default)
rtxc.live = "KG2"
try:
driver = GraphDatabase.driver(rtxc.neo4j_bolt, auth=(rtxc.neo4j_username, rtxc.neo4j_password))
with driver.session() as session:
query_results = session.run(cypher_query).data()
driver.close()
except Exception:
tb = traceback.format_exc()
error_type, error, _ = sys.exc_info()
log.error(f"Encountered an error interacting with {kp} neo4j. {tb}", error_code=error_type.__name__)
return []
else:
return query_results
encrypted: bool = False,
validate_ssl: bool = False) -> None:
"""
There's currently no request timeout from client side where server
side can be enforced via "dbms.transaction.timeout"
By default, it will set max number of connections to 50 and connection time out to 10 seconds.
:param endpoint: neo4j endpoint
:param num_conns: number of connections
:param max_connection_lifetime_sec: max life time the connection can have when it comes to reuse. In other
words, connection life time longer than this value won't be reused and closed on garbage collection. This
value needs to be smaller than surrounding network environment's timeout.
"""
endpoint = f'{host}:{port}'
LOGGER.info('NEO4J endpoint: {}'.format(endpoint))
trust = neo4j.TRUST_SYSTEM_CA_SIGNED_CERTIFICATES if validate_ssl else neo4j.TRUST_ALL_CERTIFICATES
self._driver = GraphDatabase.driver(endpoint, max_connection_pool_size=num_conns,
connection_timeout=10,
max_connection_lifetime=max_connection_lifetime_sec,
auth=(user, password),
encrypted=encrypted,
trust=trust) # type: Driver