Kivy2Neo sending cypher queries from a Kivy app to a Neo4j database

Kivy2Neo sending cypher queries from a Kivy app to a Neo4j database

  • @Tsartsaris

    Posted on 2014-10-05

        Recently I was working on an Android app for a small Real Estate company. Not being a Java developer can be a pain in the a$$, especially when you have to develop such an app. I wouldn't take the job if the Kivy Framework didn't exist. 

        As almost in every project, it comes to a point that you have to connect to a database. In my case I had to connect to my Neo4j database save some data, retrieve some records and so on.

        ​While I could use some of the existing libraries to connect from Kivy and Python to the Neo4j database, there was a simple goal to achieve: keep the app as lightweight as  possible. So I started coding a really simple class, that could handle such a functionality. The outcome is not superb but it does the job. Minimizing at the lowest possibly level the libraries to import and use , I used the to handle the transaction exploiting the /data/transaction/commit HTTP endpoint of Neo4j. 

    ​    Using the class at the most simple way is as easy as 

    from import App
    from kivy.uix.label import Label
    from kivy2neo import *
    class MyApp(App):
    	neo = KivyConnNeo()
    	neo.cypher('MATCH (n:User) RETURN n')
    	print neo.results
    	def build(self):
    		return Label(text='')
    if __name__ == '__main__':

    ​    To change the database just use a parameter in the KivyConnNeo section like so 

    	neo = KivyConnNeo('')

    ​    The is a lot of work to be done like error checking, different HHTP endpoints for cypher queries etc. This is just a starting point that simply sends a cypher query and reads the response. 

    ​    You can find the code below or on GitHub 

    from import UrlRequest
    import json
    headers = {'Content-type': 'application/json',
    		'Accept': 'application/json; charset=UTF-8'}
    class KivyConnNeo(object):
    	def __init__(self,base_url=""):
    		self.base_url = base_url
    		self.results = []
    	def post_query(self,req,result):
    		#print json.dumps(result,indent=4, sort_keys=True)
    		if result:
    			for key, value in result.iteritems():
    				if key == 'errors':
    					for s in value:
    						print s['message']
    				elif key == 'results':
    					for s in value:
    						for t in s['data']:
    							for e in t['row']:
    	def print_error(self,req,error):
    		print type(error)
    		print error
    	def cypher(self, query):
    		self.query = query
    		self.params = json.dumps({
    		"statements" : [ {
    		"statement" : ""+self.query+"",
    		"resultDataContents" : [ "row", "graph" ]
    		} ]
    		self.req = UrlRequest(self.base_url + '/db/data/transaction/commit',
    			on_success=self.post_query ,req_headers=headers, on_error=self.print_error, req_body = self.params , debug = True, method = 'POST')



Social Me!

Twitter Logo LinkedIn Logo Google+ Logo Tumblr Logo