Skip to content

Instantly share code, notes, and snippets.

@arlindosilvaneto
Last active October 19, 2016 15:45
Show Gist options
  • Save arlindosilvaneto/216bbc066ea8e8832024c006d1ae43b7 to your computer and use it in GitHub Desktop.
Save arlindosilvaneto/216bbc066ea8e8832024c006d1ae43b7 to your computer and use it in GitHub Desktop.
Neo4j Python Decorator Handler

Neo4j Decorator Driver for Python

All annotations assumes class methods as decorated sources.

Query Decorator Signature

@Query([cypher statement], asTable=False)
  • asTable: If true, the method will get the result parameter as a DataFrame Pandas object

Transactional Decorator Signature

@Transactional

Minimum query decorated method signagure

def method_name(self, result, *args):
    ...

Extra args passed could be

  • tx: A transaction to be used (could be the one injected with Transactional decorator)
  • parameters: A dictionary with statement parameters

Minimum Transactional decorated method signature

def method_name(self, tx):
    ...
import sys
from db.neo4j import Transactional, Query
class UserDao():
@Query("match (user:USER) return user")
def searchUsers(self, result, *args):
user = result.next()
return user
@Query("match (a)-[r]->(b) where {label} in labels(a) return r", asTable=True)
def searchUsersConnections(self, result, *args):
print(result['r'][0])
class App():
def __init__(self):
self.dao = UserDao()
def checkUsers(self):
self.call()
def checkConnections(self):
self.dao.searchUsersConnections(parameters={"label":"USER"})
@Transactional
def call(self, tx):
users = self.dao.searchUsers(tx=tx)
if __name__ == "__main__":
app = App()
app.checkUsers()
app.checkConnections()
from py2neo import Graph
# open database connection
class Neo4j():
connection = None
@staticmethod
def graph():
if Neo4j.connection is None:
print("Opening Database Connection")
Neo4j.connection = Graph(host="<url>", user="neo4j", password="<password>")
return Neo4j.connection
# decorator to inject transaction into a class method
def Transactional(method):
def openTransaction(self=None, tx=None):
# call method with transaction
transaction = tx or Neo4j.graph().begin()
result = None
try:
result = method(self=self, tx = transaction)
transaction.commit()
except:
transaction.rollback()
raise
return result
return openTransaction
# decorator to run query over a given or injected transaction
def Query(statement, asTable=False):
def neo4jQuery(method):
def startTxAndQuery(self, tx=Neo4j.graph(), parameters=None):
print("Transaction", tx)
result = tx.run(statement, parameters=parameters)
if asTable:
result = DataFrame(result.data())
return method(self, result=result)
return startTxAndQuery
return neo4jQuery
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment