Neo4j Python Decorator Handler

README.md

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):
    ...

app.py

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()

neo4j.py

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