Py2neo - How To Create a Neo4j Graph Object Using The Py2neo Framework

This article is about using the Py2neo framework to access Neo4j.  A simple test UI for connecting to Neo4j databases is provided written in python 3.x using the V3 version of Py2neo.  The documentation for the Graph class can be found here.  The UI will allow you to modify any of the parameters used when creating a "graph" object.  Here is a snippet of code taken from the documentation:

>>> from py2neo import Graph
>>> graph_1 = Graph()
>>> graph_2 = Graph(host="localhost")
>>> graph_3 = Graph("http://localhost:7474/db/data/")

This shows several examples of creating a Graph object all of which produce the same result. You have two choices - specify a URI for the Neo4j instance (see graph_3) or specify the keyword parameters directly.  All of the keywords have defaults except "password" so graph_1 will work assuming you have security disabled.  The test UI provided below takes the approach of supplying the values for the keyword parameters and then showing you the resulting URI created by Py2neo.  Only the keywords you supply values for are given to the Graph function.  The keyword parameters and their defaults are shown below (from the documentation):

 

Creating the graph object will cause an http get operation on the Neo4j Server.  A number of things can go wrong from bad port number to bad userid/password. The program then runs a simple cypher command ("match (n) return n limit 1") which will use HTTP or Bolt depending on how you set the parameters. The program displays any errors and shows the HTTP and Bolt streams.  

The code below was developed using Python 3.5 and the V3 version of Py2neo.  It uses the tkinter UI framework installed with Python so I think it will work on all supported platforms but I've only run it on Windows 10.  This might work with earlier versions of Python but you must have the V3 version of Py2neo installed.  Also, I'm not convinced the http stream is updating correctly in the UI, still working on that...  :) 

#!/usr/bin/env python3
from io import StringIO
from py2neo import Graph
from py2neo import DBMS
from py2neo import watch
from tkinter import *
from tkinter import messagebox
import tkinter.scrolledtext as tkst
  
class Application(Frame):
    def __init__(self, master):     
        super(Application,self).__init__(master)
        self.httpstream = StringIO("")
        self.boltstream = StringIO("")
        watch("neo4j.bolt",out=self.boltstream)
        watch("neo4j.http", out=self.httpstream)
        self.myGraph = None
        self.grid()
        self.bolt = None
        self.autodetect = BooleanVar()
        self.usebolt = BooleanVar()
        self.securecon = BooleanVar()
        self.createWidgets()
        
    def createWidgets(self):
        # autodetect
        gridrow=1
        self.chkbox_autodetect = Checkbutton(self, text = "Autodetect BOLT", variable=self.autodetect, command=self.chkbox_autodetect_Click)
        self.chkbox_autodetect.select()
        self.chkbox_autodetect.grid(row=gridrow,column=0,sticky=W)
        # use Bolt
        gridrow=2
        self.chkbox_usebolt = Checkbutton(self, text = "Use BOLT", variable=self.usebolt, command=self.chkbox_usebolt_Click)
        self.chkbox_usebolt.deselect()
        self.chkbox_usebolt.grid(row=gridrow,column=0,sticky=W)
        # use secure connection
        gridrow=3
        self.chkbox_securecon = Checkbutton(self, text = "Use secure connection", variable=self.securecon, command=self.chkbox_securecon_Click)
        self.chkbox_securecon.grid(row=gridrow,column=0,sticky=W)        
        # host
        gridrow =gridrow + 1
        self.lbl2 = Label(self, text="Host :")
        self.lbl2.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_host = Entry(self)
        self.txtbox_host.insert(0,"localhost")
        self.txtbox_host.grid(row=gridrow, column=1, sticky = W)
        # http port
        gridrow =gridrow + 1
        self.lbl3 = Label(self, text="HTTP Port :")
        self.lbl3.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_httpport = Entry(self)
        self.txtbox_httpport.insert(0,"7474")
        self.txtbox_httpport.grid(row=gridrow, column=1, sticky = W)
        # https port
        gridrow =gridrow + 1
        self.lbl4 = Label(self, text="HTTPS Port :")
        self.lbl4.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_httpsport = Entry(self)
        self.txtbox_httpsport.insert(0,"7473")
        self.txtbox_httpsport.grid(row=gridrow, column=1, sticky = W)
        
        # BOLT port
        gridrow =gridrow + 1
        self.lbl5 = Label(self, text="BOLT Port :")
        self.lbl5.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_boltport = Entry(self)
        self.txtbox_boltport.insert(0,"7687")
        self.txtbox_boltport.grid(row=gridrow, column=1, sticky = W)
        # userid
        gridrow =gridrow + 1
        self.lbl6 = Label(self, text="Userid :")
        self.lbl6.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_userid = Entry(self)
        self.txtbox_userid.insert(0,"neo4j")
        self.txtbox_userid.grid(row=gridrow, column=1, sticky = W)
        # password
        gridrow =gridrow + 1
        self.lbl7 = Label(self, text="Password :")
        self.lbl7.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_password = Entry(self)
        self.txtbox_password.grid(row=gridrow, column=1, sticky = W)
        gridrow =gridrow + 1
        self.cmdbtn_Connect = Button(self,text="Connect",command=self.cmdbtn_Connect_Click)
        self.cmdbtn_Connect.grid(row=gridrow, column=0, sticky = W)        
        #output
        gridrow =gridrow + 1
        self.lbl10 = Label(self, text="keyword arguments used to create Graph object :")
        self.lbl10.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_kwargs = Text(self,width=80, height = 1, wrap = WORD)
        self.txtbox_kwargs.grid(row=gridrow, column=1, sticky=W)        
        gridrow =gridrow + 1
        self.lbl8 = Label(self, text="graph.dbms :")
        self.lbl8.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_dbms = Text(self,width=50, height = 1, wrap = WORD)
        self.txtbox_dbms.grid(row=gridrow, column=1, sticky=W)
        #http stream
        gridrow =gridrow + 1
        self.lbl9 = Label(self, text="http stream :")
        self.lbl9.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_http = tkst.ScrolledText(self,width=50, height = 5, wrap = WORD)
        self.txtbox_http.grid(row=gridrow, column=1, sticky=W)
        #bolt stream
        gridrow =gridrow + 1
        self.lbl9 = Label(self, text="bolt stream :")
        self.lbl9.grid(row=gridrow, column=0,sticky=W)
        self.txtbox_bolt = tkst.ScrolledText(self,width=50, height = 5, wrap = WORD)
        self.txtbox_bolt.grid(row=gridrow, column=1, sticky=W)
        
    def chkbox_autodetect_Click(self):
        if self.autodetect.get():
            self.chkbox_usebolt.configure(state='disable')
        else:
            self.chkbox_usebolt.configure(state='normal')
            
    def chkbox_usebolt_Click(self):
            pass
    def chkbox_securecon_Click(self):
            pass
    
        
    def clear_output(self, ):
        self.txtbox_kwargs.delete(0.0,END)
        self.txtbox_dbms.delete(0.0,END)
        self.txtbox_http.delete(0.0,END)
        self.txtbox_bolt.delete(0.0,END)
        self.httpstream = StringIO("")
        self.boltstream = StringIO("")
        watch("neo4j.bolt",out=self.boltstream)
        watch("neo4j.http", out=self.httpstream)
        self.update()
    
    # create a graph object using the parameters from the UI and access the database    
    def cmdbtn_Connect_Click(self):
        
        self.clear_output()
        if self.autodetect.get():
            self.bolt = None
        else:
            if self.usebolt.get():
               self.bolt = True
            else:
               self.bolt = False
               
        # only add the keyword argument if the user supplied a value
        graphkwargs = {}
        graphkwargs['bolt'] = self.bolt
        graphkwargs['secure'] = self.securecon.get()
        if len(self.txtbox_host.get()) > 0:
            graphkwargs['host'] = self.txtbox_host.get()
        if len(self.txtbox_userid.get()) > 0:
            graphkwargs['user'] = self.txtbox_userid.get()
        if len( self.txtbox_password.get()) > 0:   
            graphkwargs['password'] = self.txtbox_password.get()
        try:
            self.http_port = int(self.txtbox_httpport.get())
            graphkwargs['http_port']=self.http_port
        except:
            pass
            #self.http_port = 7474
        try:
            self.https_port = int(self.txtbox_httpsport.get())
            graphkwargs['https_port']=self.https_port
        except:
            pass
            #self.https_port = 7473
        try:
            self.bolt_port = int(self.txtbox_boltport.get())
            graphkwargs['bolt_port']=self.bolt_port
        except:
            pass
            #self.bolt_port = 7587
            
        #print("keyword arguments used: {}".format(str(graphkwargs)))
        self.txtbox_kwargs.insert(0.0,str(graphkwargs))
        
        try:
            self.myGraph = None
            self.myGraph = Graph(**graphkwargs)

        except BaseException as e:
            messagebox.showerror("Create Graph Object Error", "{} - Create Graph Object Error.".format(repr(e)))
        else:    
            # get the URI generated by py2neo
            try:
                self.txtbox_dbms.delete(0.0,END)
                self.txtbox_dbms.insert(0.0,self.myGraph.dbms)
            except BaseException as e:
                messagebox.showerror("Get graph.dbms", "{} - Get graph.dbms failed.".format(repr(e)))
    
            # return one node using a cypher command
            try:
                self.myGraph.run("match (n) return n limit 1")
            except BaseException as e:
                messagebox.showerror("Run Cypher Command", "{} - Run Cypher Command failed.".format(repr(e)))
        
        #self.txtbox_http.delete(0.0,END)
        self.txtbox_http.insert(0.0,self.httpstream.getvalue())
        #self.txtbox_bolt.delete(0.0,END)
        self.txtbox_bolt.insert(0.0,self.boltstream.getvalue())
        self.update()
        
if __name__ == '__main__':        
    root = Tk()
    root.title("Neo4j Connection Tester")
    root.geometry("1000x500")   
    app = Application(root)
    root.mainloop()